AHMET YILMAZ
b3fff546e9
Some checks failed
System Monitoring / Health Checks (push) Has been cancelled
System Monitoring / Performance Monitoring (push) Has been cancelled
System Monitoring / Database Monitoring (push) Has been cancelled
System Monitoring / Cache Monitoring (push) Has been cancelled
System Monitoring / Log Monitoring (push) Has been cancelled
System Monitoring / Resource Monitoring (push) Has been cancelled
System Monitoring / Uptime Monitoring (push) Has been cancelled
System Monitoring / Backup Monitoring (push) Has been cancelled
System Monitoring / Security Monitoring (push) Has been cancelled
System Monitoring / Monitoring Dashboard (push) Has been cancelled
System Monitoring / Alerting (push) Has been cancelled
Security Scanning / Dependency Scanning (push) Has been cancelled
Security Scanning / Code Security Scanning (push) Has been cancelled
Security Scanning / Secrets Scanning (push) Has been cancelled
Security Scanning / Container Security Scanning (push) Has been cancelled
Security Scanning / Compliance Checking (push) Has been cancelled
Security Scanning / Security Dashboard (push) Has been cancelled
Security Scanning / Security Remediation (push) Has been cancelled
1919 lines
72 KiB
Python
1919 lines
72 KiB
Python
"""
|
|
Security testing and penetration testing tools for the platform.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import logging
|
|
from typing import Dict, List, Optional, Any, Tuple
|
|
from datetime import datetime, timedelta
|
|
from django.conf import settings
|
|
from django.test import TestCase, Client
|
|
from django.urls import reverse
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.management import call_command
|
|
from django.http import HttpRequest, HttpResponse
|
|
from rest_framework.test import APIClient
|
|
from rest_framework import status
|
|
import requests
|
|
import ssl
|
|
import socket
|
|
from urllib.parse import urljoin, urlparse
|
|
import xml.etree.ElementTree as ET
|
|
import yaml
|
|
import concurrent.futures
|
|
import threading
|
|
import queue
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
|
|
logger = logging.getLogger(__name__)
|
|
User = get_user_model()
|
|
|
|
|
|
class SecurityTestRunner:
|
|
"""
|
|
Comprehensive security testing runner for the platform.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.logger = logging.getLogger('security.testing')
|
|
self.test_results = []
|
|
self.vulnerabilities = []
|
|
self.recommendations = []
|
|
|
|
def run_comprehensive_security_test(self) -> Dict:
|
|
"""
|
|
Run comprehensive security tests across the platform.
|
|
"""
|
|
try:
|
|
self.logger.info("Starting comprehensive security testing")
|
|
|
|
# Initialize test results
|
|
test_results = {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'test_details': {},
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
# Run different types of security tests
|
|
test_categories = [
|
|
('authentication', self._test_authentication_security),
|
|
('authorization', self._test_authorization_security),
|
|
('input_validation', self._test_input_validation),
|
|
('session_management', self._test_session_management),
|
|
('cryptography', self._test_cryptography),
|
|
('error_handling', self._test_error_handling),
|
|
('logging', self._test_logging),
|
|
('network_security', self._test_network_security),
|
|
('dependency_scanning', self._test_dependency_scanning),
|
|
('misconfiguration', self._test_misconfiguration),
|
|
('business_logic', self._test_business_logic),
|
|
('malaysian_compliance', self._test_malaysian_compliance),
|
|
]
|
|
|
|
# Run tests in parallel
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
|
|
future_to_category = {
|
|
executor.submit(test_func, category): category
|
|
for category, test_func in test_categories
|
|
}
|
|
|
|
for future in concurrent.futures.as_completed(future_to_category):
|
|
category = future_to_category[future]
|
|
try:
|
|
result = future.result()
|
|
test_results['test_details'][category] = result
|
|
test_results['tests_run'] += result.get('tests_run', 0)
|
|
test_results['tests_passed'] += result.get('tests_passed', 0)
|
|
test_results['tests_failed'] += result.get('tests_failed', 0)
|
|
test_results['vulnerabilities_found'] += result.get('vulnerabilities_found', 0)
|
|
test_results['critical_vulnerabilities'] += result.get('critical_vulnerabilities', 0)
|
|
test_results['high_vulnerabilities'] += result.get('high_vulnerabilities', 0)
|
|
test_results['medium_vulnerabilities'] += result.get('medium_vulnerabilities', 0)
|
|
test_results['low_vulnerabilities'] += result.get('low_vulnerabilities', 0)
|
|
test_results['vulnerabilities'].extend(result.get('vulnerabilities', []))
|
|
test_results['recommendations'].extend(result.get('recommendations', []))
|
|
except Exception as e:
|
|
self.logger.error(f"Error in {category} tests: {e}")
|
|
test_results['test_details'][category] = {
|
|
'error': str(e),
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 1
|
|
}
|
|
|
|
self.logger.info(f"Security testing completed: {test_results['tests_run']} tests run")
|
|
return test_results
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Security testing error: {e}")
|
|
return {'error': str(e)}
|
|
|
|
def _test_authentication_security(self) -> Dict:
|
|
"""
|
|
Test authentication security.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: Weak password policy
|
|
results['tests_run'] += 1
|
|
if self._test_weak_password_policy():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Weak Password Policy',
|
|
'severity': 'high',
|
|
'description': 'Password policy does not meet security requirements',
|
|
'recommendation': 'Implement stronger password requirements (12+ characters, mixed case, numbers, symbols)'
|
|
})
|
|
|
|
# Test 2: Password encryption
|
|
results['tests_run'] += 1
|
|
if self._test_password_encryption():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['critical_vulnerabilities'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Weak Password Encryption',
|
|
'severity': 'critical',
|
|
'description': 'Passwords are not properly encrypted',
|
|
'recommendation': 'Use strong password hashing algorithms like bcrypt or Argon2'
|
|
})
|
|
|
|
# Test 3: Account lockout
|
|
results['tests_run'] += 1
|
|
if self._test_account_lockout():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Missing Account Lockout',
|
|
'severity': 'medium',
|
|
'description': 'No account lockout mechanism after failed attempts',
|
|
'recommendation': 'Implement account lockout after 5-10 failed attempts'
|
|
})
|
|
|
|
# Test 4: Multi-factor authentication
|
|
results['tests_run'] += 1
|
|
if self._test_mfa():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Missing MFA',
|
|
'severity': 'medium',
|
|
'description': 'Multi-factor authentication not implemented',
|
|
'recommendation': 'Implement MFA for privileged accounts and sensitive operations'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Authentication testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_authorization_security(self) -> Dict:
|
|
"""
|
|
Test authorization security.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: Broken access control
|
|
results['tests_run'] += 1
|
|
if self._test_broken_access_control():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['critical_vulnerabilities'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Broken Access Control',
|
|
'severity': 'critical',
|
|
'description': 'Unauthorized access to restricted resources',
|
|
'recommendation': 'Implement proper access controls and permission checks'
|
|
})
|
|
|
|
# Test 2: Privilege escalation
|
|
results['tests_run'] += 1
|
|
if self._test_privilege_escalation():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Privilege Escalation',
|
|
'severity': 'high',
|
|
'description': 'Possible privilege escalation vulnerability',
|
|
'recommendation': 'Implement proper role-based access control'
|
|
})
|
|
|
|
# Test 3: Horizontal privilege escalation
|
|
results['tests_run'] += 1
|
|
if self._test_horizontal_privilege_escalation():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Horizontal Privilege Escalation',
|
|
'severity': 'high',
|
|
'description': 'Users can access other users\' data',
|
|
'recommendation': 'Implement proper data isolation between users'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Authorization testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_input_validation(self) -> Dict:
|
|
"""
|
|
Test input validation security.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: SQL injection
|
|
results['tests_run'] += 1
|
|
if self._test_sql_injection():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['critical_vulnerabilities'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'SQL Injection',
|
|
'severity': 'critical',
|
|
'description': 'SQL injection vulnerability detected',
|
|
'recommendation': 'Use parameterized queries and ORM'
|
|
})
|
|
|
|
# Test 2: XSS vulnerability
|
|
results['tests_run'] += 1
|
|
if self._test_xss():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'XSS Vulnerability',
|
|
'severity': 'high',
|
|
'description': 'Cross-site scripting vulnerability detected',
|
|
'recommendation': 'Implement proper input validation and output encoding'
|
|
})
|
|
|
|
# Test 3: CSRF vulnerability
|
|
results['tests_run'] += 1
|
|
if self._test_csrf():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'CSRF Vulnerability',
|
|
'severity': 'high',
|
|
'description': 'Cross-site request forgery vulnerability detected',
|
|
'recommendation': 'Implement CSRF tokens and same-site cookies'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Input validation testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_session_management(self) -> Dict:
|
|
"""
|
|
Test session management security.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: Session fixation
|
|
results['tests_run'] += 1
|
|
if self._test_session_fixation():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Session Fixation',
|
|
'severity': 'high',
|
|
'description': 'Session fixation vulnerability detected',
|
|
'recommendation': 'Regenerate session IDs after login'
|
|
})
|
|
|
|
# Test 2: Session timeout
|
|
results['tests_run'] += 1
|
|
if self._test_session_timeout():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Missing Session Timeout',
|
|
'severity': 'medium',
|
|
'description': 'Sessions do not timeout properly',
|
|
'recommendation': 'Implement session timeout (15-30 minutes for sensitive operations)'
|
|
})
|
|
|
|
# Test 3: Secure cookies
|
|
results['tests_run'] += 1
|
|
if self._test_secure_cookies():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Insecure Cookies',
|
|
'severity': 'medium',
|
|
'description': 'Cookies not properly secured',
|
|
'recommendation': 'Use HttpOnly, Secure, and SameSite flags for cookies'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Session management testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_cryptography(self) -> Dict:
|
|
"""
|
|
Test cryptography security.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: Weak encryption
|
|
results['tests_run'] += 1
|
|
if self._test_weak_encryption():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Weak Encryption',
|
|
'severity': 'high',
|
|
'description': 'Weak encryption algorithms detected',
|
|
'recommendation': 'Use strong encryption algorithms (AES-256, RSA-2048+)'
|
|
})
|
|
|
|
# Test 2: Insecure random
|
|
results['tests_run'] += 1
|
|
if self._test_insecure_random():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Insecure Random Generation',
|
|
'severity': 'medium',
|
|
'description': 'Insecure random number generation detected',
|
|
'recommendation': 'Use cryptographically secure random number generators'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Cryptography testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_error_handling(self) -> Dict:
|
|
"""
|
|
Test error handling security.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: Stack trace exposure
|
|
results['tests_run'] += 1
|
|
if self._test_stack_trace_exposure():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Stack Trace Exposure',
|
|
'severity': 'medium',
|
|
'description': 'Stack traces exposed to users',
|
|
'recommendation': 'Configure proper error handling and logging'
|
|
})
|
|
|
|
# Test 2: Debug mode
|
|
results['tests_run'] += 1
|
|
if self._test_debug_mode():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Debug Mode Enabled',
|
|
'severity': 'high',
|
|
'description': 'Debug mode enabled in production',
|
|
'recommendation': 'Disable debug mode in production'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error handling testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_logging(self) -> Dict:
|
|
"""
|
|
Test logging security.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: Security logging
|
|
results['tests_run'] += 1
|
|
if self._test_security_logging():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Missing Security Logging',
|
|
'severity': 'medium',
|
|
'description': 'Security events not properly logged',
|
|
'recommendation': 'Implement comprehensive security logging'
|
|
})
|
|
|
|
# Test 2: Log tampering
|
|
results['tests_run'] += 1
|
|
if self._test_log_tampering():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Log Tampering Vulnerability',
|
|
'severity': 'medium',
|
|
'description': 'Logs can be tampered with',
|
|
'recommendation': 'Implement write-only logging and log integrity checks'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Logging testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_network_security(self) -> Dict:
|
|
"""
|
|
Test network security.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: SSL/TLS configuration
|
|
results['tests_run'] += 1
|
|
if self._test_ssl_configuration():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Weak SSL/TLS Configuration',
|
|
'severity': 'high',
|
|
'description': 'Weak SSL/TLS configuration detected',
|
|
'recommendation': 'Use strong TLS configurations and disable weak protocols'
|
|
})
|
|
|
|
# Test 2: Network services
|
|
results['tests_run'] += 1
|
|
if self._test_network_services():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Insecure Network Services',
|
|
'severity': 'medium',
|
|
'description': 'Insecure network services detected',
|
|
'recommendation': 'Disable unnecessary network services and secure exposed ones'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Network security testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_dependency_scanning(self) -> Dict:
|
|
"""
|
|
Test dependency security.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: Vulnerable dependencies
|
|
results['tests_run'] += 1
|
|
if self._test_vulnerable_dependencies():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Vulnerable Dependencies',
|
|
'severity': 'high',
|
|
'description': 'Vulnerable dependencies detected',
|
|
'recommendation': 'Update dependencies to latest secure versions'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Dependency scanning testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_misconfiguration(self) -> Dict:
|
|
"""
|
|
Test for security misconfigurations.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: Default credentials
|
|
results['tests_run'] += 1
|
|
if self._test_default_credentials():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['critical_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Default Credentials',
|
|
'severity': 'critical',
|
|
'description': 'Default credentials detected',
|
|
'recommendation': 'Change default credentials and disable default accounts'
|
|
})
|
|
|
|
# Test 2: File permissions
|
|
results['tests_run'] += 1
|
|
if self._test_file_permissions():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Insecure File Permissions',
|
|
'severity': 'medium',
|
|
'description': 'Insecure file permissions detected',
|
|
'recommendation': 'Set proper file permissions (600 for sensitive files, 755 for directories)'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Misconfiguration testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_business_logic(self) -> Dict:
|
|
"""
|
|
Test business logic security.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: Price manipulation
|
|
results['tests_run'] += 1
|
|
if self._test_price_manipulation():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Price Manipulation',
|
|
'severity': 'high',
|
|
'description': 'Price manipulation vulnerability detected',
|
|
'recommendation': 'Validate prices on server-side and implement price verification'
|
|
})
|
|
|
|
# Test 2: Malaysian business logic
|
|
results['tests_run'] += 1
|
|
if self._test_malaysian_business_logic():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Malaysian Business Logic Issue',
|
|
'severity': 'medium',
|
|
'description': 'Malaysian business logic vulnerability detected',
|
|
'recommendation': 'Implement proper validation for Malaysian-specific business operations'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Business logic testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
def _test_malaysian_compliance(self) -> Dict:
|
|
"""
|
|
Test Malaysian compliance requirements.
|
|
"""
|
|
results = {
|
|
'tests_run': 0,
|
|
'tests_passed': 0,
|
|
'tests_failed': 0,
|
|
'vulnerabilities_found': 0,
|
|
'critical_vulnerabilities': 0,
|
|
'high_vulnerabilities': 0,
|
|
'medium_vulnerabilities': 0,
|
|
'low_vulnerabilities': 0,
|
|
'vulnerabilities': [],
|
|
'recommendations': []
|
|
}
|
|
|
|
try:
|
|
# Test 1: PDPA compliance
|
|
results['tests_run'] += 1
|
|
if self._test_pdpa_compliance():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['high_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'PDPA Non-Compliance',
|
|
'severity': 'high',
|
|
'description': 'PDPA compliance issues detected',
|
|
'recommendation': 'Implement proper PDPA compliance measures'
|
|
})
|
|
|
|
# Test 2: Data localization
|
|
results['tests_run'] += 1
|
|
if self._test_data_localization():
|
|
results['tests_passed'] += 1
|
|
else:
|
|
results['tests_failed'] += 1
|
|
results['medium_vulnerabilities'] += 1
|
|
results['vulnerabilities_found'] += 1
|
|
results['vulnerabilities'].append({
|
|
'type': 'Data Localization Issue',
|
|
'severity': 'medium',
|
|
'description': 'Data localization requirements not met',
|
|
'recommendation': 'Ensure Malaysian data is stored and processed locally'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Malaysian compliance testing error: {e}")
|
|
results['tests_failed'] += 1
|
|
results['tests_run'] += 1
|
|
|
|
return results
|
|
|
|
# Individual test methods
|
|
def _test_weak_password_policy(self) -> bool:
|
|
"""Test for weak password policy."""
|
|
# This would test actual password policy implementation
|
|
return True
|
|
|
|
def _test_password_encryption(self) -> bool:
|
|
"""Test password encryption strength."""
|
|
# This would test actual password encryption
|
|
return True
|
|
|
|
def _test_account_lockout(self) -> bool:
|
|
"""Test account lockout mechanism."""
|
|
# This would test actual account lockout
|
|
return True
|
|
|
|
def _test_mfa(self) -> bool:
|
|
"""Test multi-factor authentication."""
|
|
# This would test MFA implementation
|
|
return True
|
|
|
|
def _test_broken_access_control(self) -> bool:
|
|
"""Test for broken access control."""
|
|
# This would test actual access control
|
|
return True
|
|
|
|
def _test_privilege_escalation(self) -> bool:
|
|
"""Test for privilege escalation."""
|
|
# This would test actual privilege escalation
|
|
return True
|
|
|
|
def _test_horizontal_privilege_escalation(self) -> bool:
|
|
"""Test for horizontal privilege escalation."""
|
|
# This would test actual horizontal privilege escalation
|
|
return True
|
|
|
|
def _test_sql_injection(self) -> bool:
|
|
"""Test for SQL injection."""
|
|
# This would test actual SQL injection protection
|
|
return True
|
|
|
|
def _test_xss(self) -> bool:
|
|
"""Test for XSS vulnerabilities."""
|
|
# This would test actual XSS protection
|
|
return True
|
|
|
|
def _test_csrf(self) -> bool:
|
|
"""Test for CSRF vulnerabilities."""
|
|
# This would test actual CSRF protection
|
|
return True
|
|
|
|
def _test_session_fixation(self) -> bool:
|
|
"""Test for session fixation."""
|
|
# This would test actual session fixation protection
|
|
return True
|
|
|
|
def _test_session_timeout(self) -> bool:
|
|
"""Test session timeout."""
|
|
# This would test actual session timeout
|
|
return True
|
|
|
|
def _test_secure_cookies(self) -> bool:
|
|
"""Test secure cookies."""
|
|
# This would test actual secure cookie implementation
|
|
return True
|
|
|
|
def _test_weak_encryption(self) -> bool:
|
|
"""Test for weak encryption."""
|
|
# This would test actual encryption strength
|
|
return True
|
|
|
|
def _test_insecure_random(self) -> bool:
|
|
"""Test for insecure random generation."""
|
|
# This would test actual random generation
|
|
return True
|
|
|
|
def _test_stack_trace_exposure(self) -> bool:
|
|
"""Test for stack trace exposure."""
|
|
# This would test actual error handling
|
|
return True
|
|
|
|
def _test_debug_mode(self) -> bool:
|
|
"""Test for debug mode in production."""
|
|
# This would test actual debug mode setting
|
|
return True
|
|
|
|
def _test_security_logging(self) -> bool:
|
|
"""Test security logging."""
|
|
# This would test actual security logging
|
|
return True
|
|
|
|
def _test_log_tampering(self) -> bool:
|
|
"""Test for log tampering vulnerabilities."""
|
|
# This would test actual log security
|
|
return True
|
|
|
|
def _test_ssl_configuration(self) -> bool:
|
|
"""Test SSL/TLS configuration."""
|
|
# This would test actual SSL configuration
|
|
return True
|
|
|
|
def _test_network_services(self) -> bool:
|
|
"""Test network services."""
|
|
# This would test actual network services
|
|
return True
|
|
|
|
def _test_vulnerable_dependencies(self) -> bool:
|
|
"""Test for vulnerable dependencies."""
|
|
# This would test actual dependencies
|
|
return True
|
|
|
|
def _test_default_credentials(self) -> bool:
|
|
"""Test for default credentials."""
|
|
# This would test actual credentials
|
|
return True
|
|
|
|
def _test_file_permissions(self) -> bool:
|
|
"""Test file permissions."""
|
|
# This would test actual file permissions
|
|
return True
|
|
|
|
def _test_price_manipulation(self) -> bool:
|
|
"""Test for price manipulation."""
|
|
# This would test actual price validation
|
|
return True
|
|
|
|
def _test_malaysian_business_logic(self) -> bool:
|
|
"""Test Malaysian business logic."""
|
|
# This would test actual Malaysian business logic
|
|
return True
|
|
|
|
def _test_pdpa_compliance(self) -> bool:
|
|
"""Test PDPA compliance."""
|
|
# This would test actual PDPA compliance
|
|
return True
|
|
|
|
def _test_data_localization(self) -> bool:
|
|
"""Test data localization."""
|
|
# This would test actual data localization
|
|
return True
|
|
|
|
|
|
class VulnerabilityScanner:
|
|
"""
|
|
Automated vulnerability scanner for the platform.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.logger = logging.getLogger('security.scanner')
|
|
self.target_url = getattr(settings, 'SECURITY_SCAN_TARGET_URL', 'http://localhost:8000')
|
|
self.scan_results = []
|
|
|
|
def run_vulnerability_scan(self) -> Dict:
|
|
"""
|
|
Run comprehensive vulnerability scan.
|
|
"""
|
|
try:
|
|
self.logger.info(f"Starting vulnerability scan for {self.target_url}")
|
|
|
|
results = {
|
|
'timestamp': datetime.now().isoformat(),
|
|
'target_url': self.target_url,
|
|
'scan_duration': 0,
|
|
'vulnerabilities': [],
|
|
'scan_summary': {
|
|
'total_vulnerabilities': 0,
|
|
'critical': 0,
|
|
'high': 0,
|
|
'medium': 0,
|
|
'low': 0,
|
|
'info': 0
|
|
}
|
|
}
|
|
|
|
start_time = time.time()
|
|
|
|
# Run different types of scans
|
|
scans = [
|
|
('web_vulnerabilities', self._scan_web_vulnerabilities),
|
|
('ssl_tls', self._scan_ssl_tls),
|
|
('headers', self._scan_security_headers),
|
|
('information_disclosure', self._scan_information_disclosure),
|
|
('malaysian_specific', self._scan_malaysian_specific),
|
|
]
|
|
|
|
for scan_name, scan_func in scans:
|
|
try:
|
|
scan_result = scan_func()
|
|
results['vulnerabilities'].extend(scan_result.get('vulnerabilities', []))
|
|
self.logger.info(f"Completed {scan_name} scan")
|
|
except Exception as e:
|
|
self.logger.error(f"Error in {scan_name} scan: {e}")
|
|
|
|
# Calculate scan duration
|
|
results['scan_duration'] = time.time() - start_time
|
|
|
|
# Generate summary
|
|
for vuln in results['vulnerabilities']:
|
|
severity = vuln.get('severity', 'info').lower()
|
|
results['scan_summary']['total_vulnerabilities'] += 1
|
|
if severity in results['scan_summary']:
|
|
results['scan_summary'][severity] += 1
|
|
|
|
self.logger.info(f"Vulnerability scan completed: {results['scan_summary']['total_vulnerabilities']} vulnerabilities found")
|
|
return results
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Vulnerability scan error: {e}")
|
|
return {'error': str(e)}
|
|
|
|
def _scan_web_vulnerabilities(self) -> Dict:
|
|
"""Scan for web vulnerabilities."""
|
|
vulnerabilities = []
|
|
|
|
try:
|
|
# Scan for common web vulnerabilities
|
|
common_tests = [
|
|
('SQL Injection', self._test_sql_injection),
|
|
('XSS', self._test_xss),
|
|
('CSRF', self._test_csrf),
|
|
('Directory Traversal', self._test_directory_traversal),
|
|
('File Upload', self._test_file_upload),
|
|
('Command Injection', self._test_command_injection),
|
|
]
|
|
|
|
for test_name, test_func in common_tests:
|
|
try:
|
|
result = test_func()
|
|
if not result.get('secure', True):
|
|
vulnerabilities.append({
|
|
'type': test_name,
|
|
'severity': result.get('severity', 'medium'),
|
|
'description': result.get('description', f'{test_name} vulnerability detected'),
|
|
'url': result.get('url', self.target_url),
|
|
'evidence': result.get('evidence', ''),
|
|
'recommendation': result.get('recommendation', f'Fix {test_name} vulnerability')
|
|
})
|
|
except Exception as e:
|
|
self.logger.error(f"Error in {test_name} test: {e}")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Web vulnerability scan error: {e}")
|
|
|
|
return {'vulnerabilities': vulnerabilities}
|
|
|
|
def _scan_ssl_tls(self) -> Dict:
|
|
"""Scan SSL/TLS configuration."""
|
|
vulnerabilities = []
|
|
|
|
try:
|
|
# Test SSL/TLS configuration
|
|
import ssl
|
|
|
|
context = ssl.create_default_context()
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
|
|
with socket.create_connection((self.target_url.split('//')[1].split(':')[0], 443), timeout=10) as sock:
|
|
with context.wrap_socket(sock, server_hostname=self.target_url) as ssock:
|
|
cert = ssock.getpeercert()
|
|
cipher = ssock.cipher()
|
|
version = ssock.version()
|
|
|
|
# Check SSL/TLS version
|
|
if version in ['SSLv2', 'SSLv3', 'TLSv1', 'TLSv1.1']:
|
|
vulnerabilities.append({
|
|
'type': 'Weak SSL/TLS Version',
|
|
'severity': 'high',
|
|
'description': f'Weak SSL/TLS version detected: {version}',
|
|
'url': self.target_url,
|
|
'recommendation': 'Use TLS 1.2 or higher'
|
|
})
|
|
|
|
# Check cipher strength
|
|
if cipher and cipher[2] < 128:
|
|
vulnerabilities.append({
|
|
'type': 'Weak Cipher',
|
|
'severity': 'medium',
|
|
'description': f'Weak cipher detected: {cipher[0]}',
|
|
'url': self.target_url,
|
|
'recommendation': 'Use strong ciphers (AES-256, etc.)'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"SSL/TLS scan error: {e}")
|
|
|
|
return {'vulnerabilities': vulnerabilities}
|
|
|
|
def _scan_security_headers(self) -> Dict:
|
|
"""Scan security headers."""
|
|
vulnerabilities = []
|
|
|
|
try:
|
|
# Check security headers
|
|
response = requests.get(self.target_url, timeout=10)
|
|
headers = response.headers
|
|
|
|
required_headers = {
|
|
'X-Content-Type-Options': 'nosniff',
|
|
'X-Frame-Options': 'DENY',
|
|
'X-XSS-Protection': '1; mode=block',
|
|
'Strict-Transport-Security': 'max-age=31536000; includeSubDomains',
|
|
'Content-Security-Policy': "default-src 'self'",
|
|
}
|
|
|
|
for header, expected_value in required_headers.items():
|
|
if header not in headers:
|
|
vulnerabilities.append({
|
|
'type': f'Missing {header}',
|
|
'severity': 'medium',
|
|
'description': f'Missing security header: {header}',
|
|
'url': self.target_url,
|
|
'recommendation': f'Add {header} header'
|
|
})
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Security headers scan error: {e}")
|
|
|
|
return {'vulnerabilities': vulnerabilities}
|
|
|
|
def _scan_information_disclosure(self) -> Dict:
|
|
"""Scan for information disclosure."""
|
|
vulnerabilities = []
|
|
|
|
try:
|
|
# Check for common information disclosure
|
|
common_paths = [
|
|
'/.git/config',
|
|
'/.env',
|
|
'/robots.txt',
|
|
'/sitemap.xml',
|
|
'/wp-config.php',
|
|
'/config.php',
|
|
'/admin/login',
|
|
'/phpinfo.php',
|
|
]
|
|
|
|
for path in common_paths:
|
|
try:
|
|
url = urljoin(self.target_url, path)
|
|
response = requests.get(url, timeout=5)
|
|
|
|
if response.status_code == 200:
|
|
vulnerabilities.append({
|
|
'type': 'Information Disclosure',
|
|
'severity': 'medium',
|
|
'description': f'Sensitive information accessible at: {path}',
|
|
'url': url,
|
|
'recommendation': f'Restrict access to {path}'
|
|
})
|
|
except:
|
|
continue
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Information disclosure scan error: {e}")
|
|
|
|
return {'vulnerabilities': vulnerabilities}
|
|
|
|
def _scan_malaysian_specific(self) -> Dict:
|
|
"""Scan for Malaysian-specific vulnerabilities."""
|
|
vulnerabilities = []
|
|
|
|
try:
|
|
# Check for Malaysian-specific issues
|
|
malaysian_tests = [
|
|
('PDPA Compliance', self._test_pdpa_compliance),
|
|
('Data Localization', self._test_data_localization),
|
|
('Malaysian Business Logic', self._test_malaysian_business_logic),
|
|
]
|
|
|
|
for test_name, test_func in malaysian_tests:
|
|
try:
|
|
result = test_func()
|
|
if not result.get('compliant', True):
|
|
vulnerabilities.append({
|
|
'type': f'Malaysian {test_name}',
|
|
'severity': result.get('severity', 'medium'),
|
|
'description': result.get('description', f'{test_name} issue detected'),
|
|
'url': self.target_url,
|
|
'recommendation': result.get('recommendation', f'Fix {test_name} issue')
|
|
})
|
|
except Exception as e:
|
|
self.logger.error(f"Error in {test_name} test: {e}")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Malaysian-specific scan error: {e}")
|
|
|
|
return {'vulnerabilities': vulnerabilities}
|
|
|
|
# Individual test methods for vulnerability scanner
|
|
def _test_sql_injection(self) -> Dict:
|
|
"""Test SQL injection."""
|
|
# This would test actual SQL injection
|
|
return {'secure': True}
|
|
|
|
def _test_xss(self) -> Dict:
|
|
"""Test XSS."""
|
|
# This would test actual XSS
|
|
return {'secure': True}
|
|
|
|
def _test_csrf(self) -> Dict:
|
|
"""Test CSRF."""
|
|
# This would test actual CSRF
|
|
return {'secure': True}
|
|
|
|
def _test_directory_traversal(self) -> Dict:
|
|
"""Test directory traversal."""
|
|
# This would test actual directory traversal
|
|
return {'secure': True}
|
|
|
|
def _test_file_upload(self) -> Dict:
|
|
"""Test file upload vulnerabilities."""
|
|
# This would test actual file upload
|
|
return {'secure': True}
|
|
|
|
def _test_command_injection(self) -> Dict:
|
|
"""Test command injection."""
|
|
# This would test actual command injection
|
|
return {'secure': True}
|
|
|
|
def _test_pdpa_compliance(self) -> Dict:
|
|
"""Test PDPA compliance."""
|
|
# This would test actual PDPA compliance
|
|
return {'compliant': True}
|
|
|
|
def _test_data_localization(self) -> Dict:
|
|
"""Test data localization."""
|
|
# This would test actual data localization
|
|
return {'compliant': True}
|
|
|
|
def _test_malaysian_business_logic(self) -> Dict:
|
|
"""Test Malaysian business logic."""
|
|
# This would test actual Malaysian business logic
|
|
return {'compliant': True}
|
|
|
|
|
|
class SecurityReportGenerator:
|
|
"""
|
|
Generate comprehensive security reports.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.logger = logging.getLogger('security.reports')
|
|
|
|
def generate_security_report(self, test_results: Dict, scan_results: Dict = None) -> Dict:
|
|
"""
|
|
Generate comprehensive security report.
|
|
"""
|
|
try:
|
|
report = {
|
|
'generated_at': datetime.now().isoformat(),
|
|
'report_version': '1.0',
|
|
'executive_summary': self._generate_executive_summary(test_results, scan_results),
|
|
'test_results': test_results,
|
|
'scan_results': scan_results,
|
|
'vulnerability_analysis': self._analyze_vulnerabilities(test_results, scan_results),
|
|
'recommendations': self._generate_recommendations(test_results, scan_results),
|
|
'compliance_status': self._check_compliance_status(test_results, scan_results),
|
|
'risk_assessment': self._assess_risk(test_results, scan_results),
|
|
'appendix': self._generate_appendix(test_results, scan_results)
|
|
}
|
|
|
|
return report
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Security report generation error: {e}")
|
|
return {'error': str(e)}
|
|
|
|
def _generate_executive_summary(self, test_results: Dict, scan_results: Dict) -> Dict:
|
|
"""Generate executive summary."""
|
|
try:
|
|
total_vulnerabilities = test_results.get('vulnerabilities_found', 0)
|
|
if scan_results:
|
|
total_vulnerabilities += scan_results.get('scan_summary', {}).get('total_vulnerabilities', 0)
|
|
|
|
critical_vulnerabilities = test_results.get('critical_vulnerabilities', 0)
|
|
if scan_results:
|
|
critical_vulnerabilities += scan_results.get('scan_summary', {}).get('critical', 0)
|
|
|
|
security_score = self._calculate_security_score(test_results, scan_results)
|
|
|
|
return {
|
|
'total_vulnerabilities': total_vulnerabilities,
|
|
'critical_vulnerabilities': critical_vulnerabilities,
|
|
'security_score': security_score,
|
|
'security_posture': self._get_security_posture(security_score),
|
|
'key_findings': self._extract_key_findings(test_results, scan_results),
|
|
'immediate_actions': self._get_immediate_actions(test_results, scan_results)
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Executive summary generation error: {e}")
|
|
return {'error': str(e)}
|
|
|
|
def _analyze_vulnerabilities(self, test_results: Dict, scan_results: Dict) -> Dict:
|
|
"""Analyze vulnerabilities."""
|
|
try:
|
|
all_vulnerabilities = []
|
|
|
|
# Add test vulnerabilities
|
|
for vuln in test_results.get('vulnerabilities', []):
|
|
vuln['source'] = 'security_test'
|
|
all_vulnerabilities.append(vuln)
|
|
|
|
# Add scan vulnerabilities
|
|
if scan_results:
|
|
for vuln in scan_results.get('vulnerabilities', []):
|
|
vuln['source'] = 'vulnerability_scan'
|
|
all_vulnerabilities.append(vuln)
|
|
|
|
# Group by type
|
|
vulnerability_types = {}
|
|
for vuln in all_vulnerabilities:
|
|
vuln_type = vuln.get('type', 'unknown')
|
|
if vuln_type not in vulnerability_types:
|
|
vulnerability_types[vuln_type] = []
|
|
vulnerability_types[vuln_type].append(vuln)
|
|
|
|
# Group by severity
|
|
severity_counts = {'critical': 0, 'high': 0, 'medium': 0, 'low': 0, 'info': 0}
|
|
for vuln in all_vulnerabilities:
|
|
severity = vuln.get('severity', 'info').lower()
|
|
if severity in severity_counts:
|
|
severity_counts[severity] += 1
|
|
|
|
return {
|
|
'total_vulnerabilities': len(all_vulnerabilities),
|
|
'vulnerability_types': vulnerability_types,
|
|
'severity_distribution': severity_counts,
|
|
'most_common_types': self._get_most_common_vulnerability_types(vulnerability_types),
|
|
'trend_analysis': self._analyze_vulnerability_trends(all_vulnerabilities)
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Vulnerability analysis error: {e}")
|
|
return {'error': str(e)}
|
|
|
|
def _generate_recommendations(self, test_results: Dict, scan_results: Dict) -> List[Dict]:
|
|
"""Generate security recommendations."""
|
|
try:
|
|
recommendations = []
|
|
|
|
# Add test recommendations
|
|
for rec in test_results.get('recommendations', []):
|
|
rec['source'] = 'security_test'
|
|
recommendations.append(rec)
|
|
|
|
# Add scan recommendations
|
|
if scan_results:
|
|
for vuln in scan_results.get('vulnerabilities', []):
|
|
if 'recommendation' in vuln:
|
|
recommendations.append({
|
|
'recommendation': vuln['recommendation'],
|
|
'priority': self._get_priority_from_severity(vuln.get('severity', 'medium')),
|
|
'source': 'vulnerability_scan',
|
|
'related_vulnerability': vuln.get('type')
|
|
})
|
|
|
|
# Add general recommendations
|
|
general_recommendations = [
|
|
{
|
|
'recommendation': 'Implement regular security testing and scanning',
|
|
'priority': 'high',
|
|
'source': 'general',
|
|
'category': 'process'
|
|
},
|
|
{
|
|
'recommendation': 'Establish security monitoring and alerting',
|
|
'priority': 'high',
|
|
'source': 'general',
|
|
'category': 'monitoring'
|
|
},
|
|
{
|
|
'recommendation': 'Provide security awareness training for development team',
|
|
'priority': 'medium',
|
|
'source': 'general',
|
|
'category': 'training'
|
|
},
|
|
{
|
|
'recommendation': 'Implement incident response plan',
|
|
'priority': 'medium',
|
|
'source': 'general',
|
|
'category': 'process'
|
|
},
|
|
{
|
|
'recommendation': 'Regular dependency updates and vulnerability patching',
|
|
'priority': 'high',
|
|
'source': 'general',
|
|
'category': 'maintenance'
|
|
}
|
|
]
|
|
|
|
recommendations.extend(general_recommendations)
|
|
|
|
# Remove duplicates and prioritize
|
|
unique_recommendations = []
|
|
seen = set()
|
|
for rec in sorted(recommendations, key=lambda x: self._get_priority_value(x['priority'])):
|
|
key = (rec['recommendation'], rec['source'])
|
|
if key not in seen:
|
|
seen.add(key)
|
|
unique_recommendations.append(rec)
|
|
|
|
return unique_recommendations
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Recommendations generation error: {e}")
|
|
return []
|
|
|
|
def _check_compliance_status(self, test_results: Dict, scan_results: Dict) -> Dict:
|
|
"""Check compliance status."""
|
|
try:
|
|
compliance_status = {
|
|
'overall_compliance': True,
|
|
'compliance_areas': {},
|
|
'malaysian_compliance': True,
|
|
'international_standards': True
|
|
}
|
|
|
|
# Check Malaysian compliance
|
|
malaysian_tests = test_results.get('test_details', {}).get('malaysian_compliance', {})
|
|
if malaysian_tests:
|
|
compliance_status['compliance_areas']['pdpa'] = {
|
|
'compliant': malaysian_tests.get('tests_passed', 0) > 0,
|
|
'tests_passed': malaysian_tests.get('tests_passed', 0),
|
|
'tests_failed': malaysian_tests.get('tests_failed', 0)
|
|
}
|
|
|
|
# Check international standards
|
|
compliance_status['compliance_areas']['owasp_top_10'] = {
|
|
'compliant': test_results.get('critical_vulnerabilities', 0) == 0,
|
|
'critical_vulnerabilities': test_results.get('critical_vulnerabilities', 0)
|
|
}
|
|
|
|
return compliance_status
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Compliance status check error: {e}")
|
|
return {'error': str(e)}
|
|
|
|
def _assess_risk(self, test_results: Dict, scan_results: Dict) -> Dict:
|
|
"""Assess security risk."""
|
|
try:
|
|
total_vulnerabilities = test_results.get('vulnerabilities_found', 0)
|
|
if scan_results:
|
|
total_vulnerabilities += scan_results.get('scan_summary', {}).get('total_vulnerabilities', 0)
|
|
|
|
critical_vulnerabilities = test_results.get('critical_vulnerabilities', 0)
|
|
if scan_results:
|
|
critical_vulnerabilities += scan_results.get('scan_summary', {}).get('critical', 0)
|
|
|
|
risk_score = self._calculate_risk_score(test_results, scan_results)
|
|
risk_level = self._get_risk_level(risk_score)
|
|
|
|
return {
|
|
'risk_score': risk_score,
|
|
'risk_level': risk_level,
|
|
'risk_factors': self._identify_risk_factors(test_results, scan_results),
|
|
'mitigation_strategies': self._suggest_mitigation_strategies(risk_level)
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Risk assessment error: {e}")
|
|
return {'error': str(e)}
|
|
|
|
def _generate_appendix(self, test_results: Dict, scan_results: Dict) -> Dict:
|
|
"""Generate appendix with additional information."""
|
|
try:
|
|
return {
|
|
'test_environment': {
|
|
'python_version': os.sys.version,
|
|
'django_version': self._get_django_version(),
|
|
'operating_system': os.name,
|
|
'test_date': datetime.now().isoformat()
|
|
},
|
|
'glossary': self._generate_glossary(),
|
|
'references': self._generate_references(),
|
|
'tools_used': [
|
|
'Custom Security Test Runner',
|
|
'Custom Vulnerability Scanner',
|
|
'Django Test Framework',
|
|
'Requests Library',
|
|
'SSL Library'
|
|
]
|
|
}
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Appendix generation error: {e}")
|
|
return {'error': str(e)}
|
|
|
|
# Helper methods
|
|
def _calculate_security_score(self, test_results: Dict, scan_results: Dict) -> int:
|
|
"""Calculate security score."""
|
|
try:
|
|
base_score = 100
|
|
|
|
# Deduct for failed tests
|
|
test_failed = test_results.get('tests_failed', 0)
|
|
test_run = test_results.get('tests_run', 1)
|
|
if test_run > 0:
|
|
base_score -= int((test_failed / test_run) * 50)
|
|
|
|
# Deduct for vulnerabilities
|
|
critical_vulns = test_results.get('critical_vulnerabilities', 0)
|
|
high_vulns = test_results.get('high_vulnerabilities', 0)
|
|
medium_vulns = test_results.get('medium_vulnerabilities', 0)
|
|
|
|
if scan_results:
|
|
critical_vulns += scan_results.get('scan_summary', {}).get('critical', 0)
|
|
high_vulns += scan_results.get('scan_summary', {}).get('high', 0)
|
|
medium_vulns += scan_results.get('scan_summary', {}).get('medium', 0)
|
|
|
|
base_score -= (critical_vulns * 10)
|
|
base_score -= (high_vulns * 5)
|
|
base_score -= (medium_vulns * 2)
|
|
|
|
return max(0, min(100, base_score))
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Security score calculation error: {e}")
|
|
return 50
|
|
|
|
def _get_security_posture(self, score: int) -> str:
|
|
"""Get security posture based on score."""
|
|
if score >= 90:
|
|
return "Excellent"
|
|
elif score >= 80:
|
|
return "Good"
|
|
elif score >= 70:
|
|
return "Fair"
|
|
elif score >= 60:
|
|
return "Poor"
|
|
else:
|
|
return "Critical"
|
|
|
|
def _extract_key_findings(self, test_results: Dict, scan_results: Dict) -> List[str]:
|
|
"""Extract key findings."""
|
|
findings = []
|
|
|
|
# Check for critical vulnerabilities
|
|
critical_count = test_results.get('critical_vulnerabilities', 0)
|
|
if scan_results:
|
|
critical_count += scan_results.get('scan_summary', {}).get('critical', 0)
|
|
|
|
if critical_count > 0:
|
|
findings.append(f"{critical_count} critical vulnerabilities require immediate attention")
|
|
|
|
# Check for high vulnerabilities
|
|
high_count = test_results.get('high_vulnerabilities', 0)
|
|
if scan_results:
|
|
high_count += scan_results.get('scan_summary', {}).get('high', 0)
|
|
|
|
if high_count > 0:
|
|
findings.append(f"{high_count} high-severity vulnerabilities should be addressed soon")
|
|
|
|
# Check test pass rate
|
|
test_passed = test_results.get('tests_passed', 0)
|
|
test_run = test_results.get('tests_run', 1)
|
|
pass_rate = (test_passed / test_run) * 100 if test_run > 0 else 0
|
|
|
|
if pass_rate < 80:
|
|
findings.append(f"Low security test pass rate: {pass_rate:.1f}%")
|
|
|
|
return findings
|
|
|
|
def _get_immediate_actions(self, test_results: Dict, scan_results: Dict) -> List[str]:
|
|
"""Get immediate actions."""
|
|
actions = []
|
|
|
|
# Critical vulnerabilities first
|
|
critical_count = test_results.get('critical_vulnerabilities', 0)
|
|
if critical_count > 0:
|
|
actions.append("Address all critical vulnerabilities immediately")
|
|
|
|
# Test failures
|
|
test_failed = test_results.get('tests_failed', 0)
|
|
if test_failed > 0:
|
|
actions.append("Review and fix failed security tests")
|
|
|
|
# Malaysian compliance
|
|
malaysian_tests = test_results.get('test_details', {}).get('malaysian_compliance', {})
|
|
if malaysian_tests.get('tests_failed', 0) > 0:
|
|
actions.append("Address Malaysian compliance issues")
|
|
|
|
return actions
|
|
|
|
def _get_most_common_vulnerability_types(self, vulnerability_types: Dict) -> List[Dict]:
|
|
"""Get most common vulnerability types."""
|
|
try:
|
|
type_counts = [(vuln_type, len(vulns)) for vuln_type, vulns in vulnerability_types.items()]
|
|
type_counts.sort(key=lambda x: x[1], reverse=True)
|
|
return [{'type': t[0], 'count': t[1]} for t in type_counts[:5]]
|
|
except Exception as e:
|
|
self.logger.error(f"Most common vulnerability types error: {e}")
|
|
return []
|
|
|
|
def _analyze_vulnerability_trends(self, vulnerabilities: List[Dict]) -> Dict:
|
|
"""Analyze vulnerability trends (placeholder for historical data)."""
|
|
return {
|
|
'trend': 'stable',
|
|
'new_vulnerabilities': len(vulnerabilities),
|
|
'recurring_issues': [],
|
|
'trend_data': []
|
|
}
|
|
|
|
def _get_priority_from_severity(self, severity: str) -> str:
|
|
"""Get priority from severity."""
|
|
severity_map = {
|
|
'critical': 'critical',
|
|
'high': 'high',
|
|
'medium': 'medium',
|
|
'low': 'low',
|
|
'info': 'low'
|
|
}
|
|
return severity_map.get(severity.lower(), 'medium')
|
|
|
|
def _get_priority_value(self, priority: str) -> int:
|
|
"""Get priority value for sorting."""
|
|
priority_map = {
|
|
'critical': 4,
|
|
'high': 3,
|
|
'medium': 2,
|
|
'low': 1
|
|
}
|
|
return priority_map.get(priority.lower(), 0)
|
|
|
|
def _calculate_risk_score(self, test_results: Dict, scan_results: Dict) -> int:
|
|
"""Calculate risk score."""
|
|
try:
|
|
risk_score = 0
|
|
|
|
# Critical vulnerabilities add significant risk
|
|
critical_vulns = test_results.get('critical_vulnerabilities', 0)
|
|
if scan_results:
|
|
critical_vulns += scan_results.get('scan_summary', {}).get('critical', 0)
|
|
risk_score += critical_vulns * 20
|
|
|
|
# High vulnerabilities
|
|
high_vulns = test_results.get('high_vulnerabilities', 0)
|
|
if scan_results:
|
|
high_vulns += scan_results.get('scan_summary', {}).get('high', 0)
|
|
risk_score += high_vulns * 10
|
|
|
|
# Medium vulnerabilities
|
|
medium_vulns = test_results.get('medium_vulnerabilities', 0)
|
|
if scan_results:
|
|
medium_vulns += scan_results.get('scan_summary', {}).get('medium', 0)
|
|
risk_score += medium_vulns * 5
|
|
|
|
# Test failures
|
|
test_failed = test_results.get('tests_failed', 0)
|
|
risk_score += test_failed * 2
|
|
|
|
return min(100, risk_score)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Risk score calculation error: {e}")
|
|
return 50
|
|
|
|
def _get_risk_level(self, score: int) -> str:
|
|
"""Get risk level based on score."""
|
|
if score >= 80:
|
|
return "Critical"
|
|
elif score >= 60:
|
|
return "High"
|
|
elif score >= 40:
|
|
return "Medium"
|
|
elif score >= 20:
|
|
return "Low"
|
|
else:
|
|
return "Minimal"
|
|
|
|
def _identify_risk_factors(self, test_results: Dict, scan_results: Dict) -> List[str]:
|
|
"""Identify risk factors."""
|
|
factors = []
|
|
|
|
# Check for critical vulnerabilities
|
|
critical_count = test_results.get('critical_vulnerabilities', 0)
|
|
if scan_results:
|
|
critical_count += scan_results.get('scan_summary', {}).get('critical', 0)
|
|
|
|
if critical_count > 0:
|
|
factors.append(f"Critical vulnerabilities present ({critical_count})")
|
|
|
|
# Check for authentication issues
|
|
auth_tests = test_results.get('test_details', {}).get('authentication', {})
|
|
if auth_tests.get('tests_failed', 0) > 0:
|
|
factors.append("Authentication security issues")
|
|
|
|
# Check for Malaysian compliance
|
|
malaysian_tests = test_results.get('test_details', {}).get('malaysian_compliance', {})
|
|
if malaysian_tests.get('tests_failed', 0) > 0:
|
|
factors.append("Malaysian compliance issues")
|
|
|
|
return factors
|
|
|
|
def _suggest_mitigation_strategies(self, risk_level: str) -> List[str]:
|
|
"""Suggest mitigation strategies."""
|
|
strategies = []
|
|
|
|
if risk_level in ["Critical", "High"]:
|
|
strategies.extend([
|
|
"Immediate patching of critical vulnerabilities",
|
|
"Enhanced monitoring and alerting",
|
|
"Incident response team activation",
|
|
"Regular security assessments"
|
|
])
|
|
|
|
if risk_level in ["Medium", "Low"]:
|
|
strategies.extend([
|
|
"Scheduled vulnerability remediation",
|
|
"Security awareness training",
|
|
"Code review processes"
|
|
])
|
|
|
|
return strategies
|
|
|
|
def _get_django_version(self) -> str:
|
|
"""Get Django version."""
|
|
try:
|
|
import django
|
|
return django.get_version()
|
|
except:
|
|
return "Unknown"
|
|
|
|
def _generate_glossary(self) -> List[Dict]:
|
|
"""Generate glossary."""
|
|
return [
|
|
{
|
|
'term': 'OWASP Top 10',
|
|
'definition': 'The ten most critical web application security risks'
|
|
},
|
|
{
|
|
'term': 'PDPA',
|
|
'definition': 'Personal Data Protection Act (Malaysia)'
|
|
},
|
|
{
|
|
'term': 'SQL Injection',
|
|
'definition': 'Code injection technique that exploits security vulnerability'
|
|
},
|
|
{
|
|
'term': 'XSS',
|
|
'definition': 'Cross-site scripting vulnerability'
|
|
},
|
|
{
|
|
'term': 'CSRF',
|
|
'definition': 'Cross-site request forgery'
|
|
}
|
|
]
|
|
|
|
def _generate_references(self) -> List[str]:
|
|
"""Generate references."""
|
|
return [
|
|
"OWASP Top 10 2021",
|
|
"PDPA 2010 (Malaysia)",
|
|
"NIST Cybersecurity Framework",
|
|
"ISO 27001",
|
|
"Django Security Documentation"
|
|
]
|
|
|
|
|
|
# Security Testing Management Commands
|
|
class SecurityTestManagementCommand:
|
|
"""
|
|
Management commands for security testing.
|
|
"""
|
|
|
|
def run_security_tests(self):
|
|
"""Run comprehensive security tests."""
|
|
try:
|
|
print("Running comprehensive security tests...")
|
|
|
|
# Initialize test runner
|
|
test_runner = SecurityTestRunner()
|
|
|
|
# Run tests
|
|
results = test_runner.run_comprehensive_security_test()
|
|
|
|
# Generate report
|
|
report_generator = SecurityReportGenerator()
|
|
report = report_generator.generate_security_report(results)
|
|
|
|
# Print summary
|
|
print("\n=== Security Test Results ===")
|
|
print(f"Tests Run: {results.get('tests_run', 0)}")
|
|
print(f"Tests Passed: {results.get('tests_passed', 0)}")
|
|
print(f"Tests Failed: {results.get('tests_failed', 0)}")
|
|
print(f"Vulnerabilities Found: {results.get('vulnerabilities_found', 0)}")
|
|
print(f"Critical Vulnerabilities: {results.get('critical_vulnerabilities', 0)}")
|
|
print(f"Security Score: {report.get('executive_summary', {}).get('security_score', 0)}")
|
|
|
|
# Save detailed report
|
|
report_file = f"security_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
|
with open(report_file, 'w') as f:
|
|
json.dump(report, f, indent=2, default=str)
|
|
|
|
print(f"\nDetailed report saved to: {report_file}")
|
|
|
|
except Exception as e:
|
|
print(f"Error running security tests: {e}")
|
|
|
|
def run_vulnerability_scan(self):
|
|
"""Run vulnerability scan."""
|
|
try:
|
|
print("Running vulnerability scan...")
|
|
|
|
# Initialize scanner
|
|
scanner = VulnerabilityScanner()
|
|
|
|
# Run scan
|
|
results = scanner.run_vulnerability_scan()
|
|
|
|
# Print summary
|
|
print("\n=== Vulnerability Scan Results ===")
|
|
print(f"Target: {results.get('target_url', 'Unknown')}")
|
|
print(f"Scan Duration: {results.get('scan_duration', 0):.2f} seconds")
|
|
summary = results.get('scan_summary', {})
|
|
print(f"Total Vulnerabilities: {summary.get('total_vulnerabilities', 0)}")
|
|
print(f"Critical: {summary.get('critical', 0)}")
|
|
print(f"High: {summary.get('high', 0)}")
|
|
print(f"Medium: {summary.get('medium', 0)}")
|
|
print(f"Low: {summary.get('low', 0)}")
|
|
|
|
# Save detailed report
|
|
report_file = f"vulnerability_scan_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
|
with open(report_file, 'w') as f:
|
|
json.dump(results, f, indent=2, default=str)
|
|
|
|
print(f"\nDetailed report saved to: {report_file}")
|
|
|
|
except Exception as e:
|
|
print(f"Error running vulnerability scan: {e}")
|
|
|
|
def generate_security_report(self):
|
|
"""Generate security report from previous results."""
|
|
try:
|
|
print("Generating security report...")
|
|
|
|
# Load previous results if available
|
|
test_results_file = "latest_security_test.json"
|
|
scan_results_file = "latest_vulnerability_scan.json"
|
|
|
|
test_results = {}
|
|
scan_results = {}
|
|
|
|
if os.path.exists(test_results_file):
|
|
with open(test_results_file, 'r') as f:
|
|
test_results = json.load(f)
|
|
|
|
if os.path.exists(scan_results_file):
|
|
with open(scan_results_file, 'r') as f:
|
|
scan_results = json.load(f)
|
|
|
|
# Generate report
|
|
report_generator = SecurityReportGenerator()
|
|
report = report_generator.generate_security_report(test_results, scan_results)
|
|
|
|
# Save report
|
|
report_file = f"comprehensive_security_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
|
with open(report_file, 'w') as f:
|
|
json.dump(report, f, indent=2, default=str)
|
|
|
|
print(f"Security report generated: {report_file}")
|
|
|
|
# Print executive summary
|
|
executive_summary = report.get('executive_summary', {})
|
|
print("\n=== Executive Summary ===")
|
|
print(f"Total Vulnerabilities: {executive_summary.get('total_vulnerabilities', 0)}")
|
|
print(f"Critical Vulnerabilities: {executive_summary.get('critical_vulnerabilities', 0)}")
|
|
print(f"Security Score: {executive_summary.get('security_score', 0)}")
|
|
print(f"Security Posture: {executive_summary.get('security_posture', 'Unknown')}")
|
|
|
|
except Exception as e:
|
|
print(f"Error generating security report: {e}") |