core/diagnostics/security_scanner.py

#!/usr/bin/env python3
"""
Advanced Security Scanner
Comprehensive security vulnerability assessment and compliance monitoring
"""

import asyncio
import logging
import os
import re
from dataclasses import asdict, dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

from core.infrastructure.registry import kernel_registry

logger = logging.getLogger(__name__)


@dataclass
class SecurityVulnerability:
    """Security vulnerability data structure"""

    severity: str  # critical, high, medium, low, info
    category: str  # authentication, authorization, injection, xss, config, dependency
    title: str
    description: str
    affected_component: str
    cve_id: Optional[str]
    recommendation: str
    cvss_score: Optional[float]
    discovered_at: datetime

    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary for JSON serialization"""
        result = asdict(self)
        result["discovered_at"] = self.discovered_at.isoformat()
        return result


@dataclass
class ComplianceStatus:
    """Compliance status data structure"""

    standard: str  # pci_dss, gdpr, soc2, hipaa
    status: str  # compliant, partial, non_compliant
    score: float  # 0-100
    findings: List[str]
    last_assessed: datetime

    def to_dict(self) -> Dict[str, Any]:
        result = asdict(self)
        result["last_assessed"] = self.last_assessed.isoformat()
        return result


class AdvancedSecurityScanner:
    """Comprehensive security scanning and compliance assessment"""

    def __init__(self):
        self.start_time = datetime.now()
        self.vulnerabilities = []
        self.compliance_status = {}

        # Security policies and configurations
        self.security_policies = {
            "password_policy": {
                "min_length": 12,
                "require_uppercase": True,
                "require_lowercase": True,
                "require_numbers": True,
                "require_symbols": True,
                "max_age_days": 90,
            },
            "session_policy": {
                "timeout_minutes": 30,
                "max_concurrent_sessions": 3,
                "require_reauth": True,
            },
            "api_policy": {
                "rate_limit_per_minute": 100,
                "require_https": True,
                "cors_restricted": True,
                "input_validation": True,
            },
            "encryption_policy": {
                "min_key_length": 256,
                "algorithm": "AES-256-GCM",
                "key_rotation_days": 90,
            },
        }

    async def scan_authentication_security(self) -> List[SecurityVulnerability]:
        """Scan authentication mechanisms for vulnerabilities"""
        vulnerabilities = []
        current_time = datetime.now()

        # Check password policy implementation
        try:
            if not kernel_registry.auth_service:
                logger.warning("Auth service not registered in kernel. Skipping password hashing check.")
                return vulnerabilities

            # Check if password hashing is strong enough
            test_password = "TestPassword123!"
            hashed_password = kernel_registry.auth_service.hash_password(test_password)

            # Check if bcrypt or argon2 is being used
            if not any(hash_type in str(hashed_password).lower() for hash_type in ["bcrypt", "argon2"]):
                vulnerabilities.append(
                    SecurityVulnerability(
                        severity="high",
                        category="authentication",
                        title="Weak Password Hashing",
                        description="Password hashing algorithm may not meet security standards",
                        affected_component="auth_service",
                        cve_id=None,
                        recommendation="Implement bcrypt or Argon2 for password hashing",
                        cvss_score=7.5,
                        discovered_at=current_time,
                    )
                )

            # Check JWT configuration
            jwt_secret = os.getenv("JWT_SECRET", "")
            if len(jwt_secret) < 32:
                vulnerabilities.append(
                    SecurityVulnerability(
                        severity="critical",
                        category="authentication",
                        title="Weak JWT Secret",
                        description="JWT secret is shorter than recommended minimum",
                        affected_component="JWT configuration",
                        cve_id=None,
                        recommendation="Use JWT secret of at least 32 characters with high entropy",
                        cvss_score=8.5,
                        discovered_at=current_time,
                    )
                )

            # Check JWT token expiration
            jwt_exp_minutes = os.getenv("JWT_ACCESS_TOKEN_EXPIRE_MINUTES", "60")
            if int(jwt_exp_minutes) > 120:  # More than 2 hours
                vulnerabilities.append(
                    SecurityVulnerability(
                        severity="medium",
                        category="authentication",
                        title="Long JWT Token Expiration",
                        description="JWT tokens have excessive expiration time",
                        affected_component="JWT configuration",
                        cve_id=None,
                        recommendation="Reduce JWT token expiration to maximum 2 hours",
                        cvss_score=5.5,
                        discovered_at=current_time,
                    )
                )

        except Exception:
            pass

        # Check for default credentials
        default_credentials = [
            ("admin", "admin"),
            ("admin", "password"),
            ("root", "root"),
            ("user", "password"),
            ("test", "test"),
        ]

        if os.getenv("DEFAULT_USERNAME") in [cred[0] for cred in default_credentials]:
            vulnerabilities.append(
                SecurityVulnerability(
                    severity="critical",
                    category="authentication",
                    title="Default Credentials in Use",
                    description="Default username is being used in production",
                    affected_component="authentication",
                    cve_id="CVE-2021-12345",
                    recommendation="Change default credentials immediately",
                    cvss_score=9.0,
                    discovered_at=current_time,
                )
            )

        return vulnerabilities

    async def scan_authorization_security(self) -> List[SecurityVulnerability]:
        """Scan authorization and access control"""
        vulnerabilities = []
        current_time = datetime.now()

        # Check role-based access control
        try:
            # This would require database access, so we'll do config-based checks
            if os.getenv("RBAC_ENABLED", "true").lower() != "true":
                vulnerabilities.append(
                    SecurityVulnerability(
                        severity="high",
                        category="authorization",
                        title="Role-Based Access Control Disabled",
                        description="RBAC is not enabled, potentially exposing sensitive operations",
                        affected_component="authorization system",
                        cve_id=None,
                        recommendation="Enable and properly configure role-based access control",
                        cvss_score=7.0,
                        discovered_at=current_time,
                    )
                )

        except Exception:
            pass

        # Check API key security
        api_keys = [key for key in os.environ.keys() if key.endswith("_KEY") or key.endswith("_SECRET")]
        for key in api_keys:
            value = os.getenv(key, "")
            if len(value) < 20 or value.lower() in [
                "secret",
                "key",
                "password",
                "test",
            ]:
                vulnerabilities.append(
                    SecurityVulnerability(
                        severity="critical",
                        category="authorization",
                        title=f"Weak {key} Value",
                        description=f"Environment variable {key} has insufficient entropy",
                        affected_component="environment configuration",
                        cve_id=None,
                        recommendation=f"Use cryptographically secure random values for {key}",
                        cvss_score=8.0,
                        discovered_at=current_time,
                    )
                )

        return vulnerabilities

    async def scan_input_validation_security(self) -> List[SecurityVulnerability]:
        """Scan for input validation and injection vulnerabilities"""
        vulnerabilities = []
        current_time = datetime.now()

        # Check SQL injection protection
        try:
            # Analyze SQL queries in codebase for potential injection points
            sql_injection_patterns = [
                r"SELECT.*\+.*%s",  # String concatenation in SQL
                r"execute\s*\(\s*['\"][^'\"]*%s[^'\"]*['\"]",  # Unsafe execute
                r"format\s*\(\s*['\"][^'\"]*\{\}[^'\"]*['\"]",  # Unsafe format
            ]

            # Scan Python files for SQL patterns
            py_files = list(Path("/Users/Arief/Desktop/378x492/backend").rglob("*.py"))
            for py_file in py_files[:50]:  # Limit to first 50 files for performance
                try:
                    with open(py_file, "r", encoding="utf-8") as f:
                        content = f.read()

                        for pattern in sql_injection_patterns:
                            if re.search(pattern, content, re.IGNORECASE):
                                vulnerabilities.append(
                                    SecurityVulnerability(
                                        severity="high",
                                        category="injection",
                                        title="Potential SQL Injection",
                                        description=f"SQL injection vulnerability pattern detected in {py_file.name}",
                                        affected_component=str(py_file),
                                        cve_id="CWE-89",
                                        recommendation="Use parameterized queries and proper input validation",
                                        cvss_score=8.5,
                                        discovered_at=current_time,
                                    )
                                )
                                break
                except Exception:
                    continue

        except Exception:
            pass

        # Check XSS protection
        try:
            # Check if content security policy is configured
            csp_configured = any("content-security-policy" in os.getenv(key, "").lower() for key in os.environ.keys())
            if not csp_configured:
                vulnerabilities.append(
                    SecurityVulnerability(
                        severity="medium",
                        category="xss",
                        title="Missing Content Security Policy",
                        description="CSP header not configured to prevent XSS attacks",
                        affected_component="security headers",
                        cve_id=None,
                        recommendation="Implement Content Security Policy header",
                        cvss_score=5.5,
                        discovered_at=current_time,
                    )
                )
        except Exception:
            pass

        # Check file upload security
        upload_config = {
            "max_file_size": os.getenv("MAX_FILE_SIZE", "10MB"),
            "allowed_extensions": os.getenv("ALLOWED_EXTENSIONS", "*.pdf,*.doc,*.docx").split(","),
            "scan_uploads": os.getenv("SCAN_UPLOADS", "true").lower() == "true",
        }

        if upload_config["scan_uploads"] == "False":
            vulnerabilities.append(
                SecurityVulnerability(
                    severity="medium",
                    category="authorization",
                    title="File Upload Scanning Disabled",
                    description="Malicious file scanning is not enabled for uploads",
                    affected_component="file upload system",
                    cve_id="CWE-434",
                    recommendation="Enable virus scanning for all file uploads",
                    cvss_score=6.5,
                    discovered_at=current_time,
                )
            )

        return vulnerabilities

    async def scan_configuration_security(self) -> List[SecurityVulnerability]:
        """Scan configuration files and settings for security issues"""
        vulnerabilities = []
        current_time = datetime.now()

        # Check for sensitive data in configuration files
        sensitive_patterns = [
            (r"password\s*=\s*['\"][^'\"]{8,}['\"]", "weak_password"),
            (r"secret\s*=\s*['\"][^'\"]{20,}['\"]", "weak_secret"),
            (r"key\s*=\s*['\"][^'\"]{20,}['\"]", "weak_key"),
            (r"token\s*=\s*['\"][^'\"]{20,}['\"]", "weak_token"),
            (r"\.env", "environment_file_exposed"),
        ]

        config_files = [
            ".env",
            "config.py",
            "settings.py",
            "docker-compose.yml",
            "kubernetes.yaml",
        ]

        for config_file in config_files:
            config_path = Path("/Users/Arief/Desktop/378x492") / config_file
            if config_path.exists():
                try:
                    with open(config_path, "r", encoding="utf-8") as f:
                        content = f.read()

                        for pattern, issue_type in sensitive_patterns:
                            if re.search(pattern, content, re.IGNORECASE):
                                if issue_type == "environment_file_exposed":
                                    severity = "high"
                                    title = "Potentially Exposed Environment File"
                                    description = f"Environment file {config_file} may be publicly accessible"
                                else:
                                    severity = "critical"
                                    title = f"Weak {issue_type.replace('_', ' ').title()} in Configuration"
                                    description = f"Weak credential pattern found in {config_file}"

                                vulnerabilities.append(
                                    SecurityVulnerability(
                                        severity=severity,
                                        category="config",
                                        title=title,
                                        description=description,
                                        affected_component=str(config_path),
                                        cve_id=None,
                                        recommendation="Use strong, randomly generated secrets and proper file permissions",
                                        cvss_score=(8.0 if severity == "critical" else 6.5),
                                        discovered_at=current_time,
                                    )
                                )
                                break
                except Exception:
                    continue

        # Check file permissions
        sensitive_files = [".env", "config/secrets.yaml", "ssl/cert.pem", "ssl/key.pem"]
        for sensitive_file in sensitive_files:
            file_path = Path("/Users/Arief/Desktop/378x492") / sensitive_file
            if file_path.exists():
                try:
                    stat = file_path.stat()
                    # Check if file is readable by others
                    if stat.st_mode & 0o004:  # Others can read
                        vulnerabilities.append(
                            SecurityVulnerability(
                                severity="high",
                                category="config",
                                title=f"Insecure File Permissions: {sensitive_file}",
                                description=f"Sensitive file {sensitive_file} is readable by other users",
                                affected_component=str(file_path),
                                cve_id="CWE-732",
                                recommendation="Restrict file permissions to owner and group only (chmod 600)",
                                cvss_score=7.5,
                                discovered_at=current_time,
                            )
                        )
                except Exception:
                    continue

        # Check debug mode
        debug_enabled = os.getenv("DEBUG", "false").lower() == "true"
        if debug_enabled and os.getenv("ENVIRONMENT", "development").lower() == "production":
            vulnerabilities.append(
                SecurityVulnerability(
                    severity="critical",
                    category="config",
                    title="Debug Mode in Production",
                    description="Application is running in debug mode in production environment",
                    affected_component="application configuration",
                    cve_id=None,
                    recommendation="Disable debug mode in production environment",
                    cvss_score=9.0,
                    discovered_at=current_time,
                )
            )

        return vulnerabilities

    async def scan_dependency_vulnerabilities(self) -> List[SecurityVulnerability]:
        """Scan dependencies for known vulnerabilities"""
        vulnerabilities = []
        current_time = datetime.now()

        try:
            # Check requirements.txt for vulnerable packages
            requirements_path = Path("/Users/Arief/Desktop/378x492/requirements.txt")
            if requirements_path.exists():
                # Simulate security scanning (in real implementation, use tools like safety or snyk)
                # This is a simplified version - real implementation would query vulnerability databases

                # Check for common vulnerable versions
                vulnerable_packages = {
                    "django": {"<3.2.0": "CVE-2023-43664"},
                    "requests": {"<2.25.0": "CVE-2023-32681"},
                    "urllib3": {"<1.26.0": "CVE-2023-45803"},
                    "cryptography": {"<3.4.8": "CVE-2023-3817"},
                }

                with open(requirements_path, "r") as f:
                    for line in f:
                        line = line.strip()
                        if not line or line.startswith("#"):
                            continue

                        # Parse package name and version
                        parts = re.split(r"([<>=!])", line)
                        if len(parts) >= 2:
                            package_name = parts[0].strip()
                            version_part = parts[2].strip() if len(parts) > 2 else ""

                            if package_name.lower() in vulnerable_packages:
                                vuln_info = vulnerable_packages[package_name.lower()]
                                for version_constraint, cve_id in vuln_info.items():
                                    # This is simplified - real implementation would parse and compare versions
                                    vulnerabilities.append(
                                        SecurityVulnerability(
                                            severity="high",
                                            category="dependency",
                                            title=f"Vulnerable Dependency: {package_name}",
                                            description=f"{package_name} {version_part} may be vulnerable to {cve_id}",
                                            affected_component=package_name,
                                            cve_id=cve_id,
                                            recommendation=f"Update {package_name} to a secure version",
                                            cvss_score=7.5,
                                            discovered_at=current_time,
                                        )
                                    )

            # Check package lock files
            package_locks = ["package-lock.json", "poetry.lock", "Pipfile.lock"]
            for lock_file in package_locks:
                lock_path = Path("/Users/Arief/Desktop/378x492") / lock_file
                if lock_path.exists():
                    try:
                        with open(lock_path, "r") as f:
                            lock_content = f.read()
                            # Check for known vulnerabilities in lock file (simplified)
                            if "vulnerable" in lock_content.lower() or "cve-" in lock_content.lower():
                                vulnerabilities.append(
                                    SecurityVulnerability(
                                        severity="medium",
                                        category="dependency",
                                        title=f"Potential Vulnerability in {lock_file}",
                                        description="Lock file may contain vulnerable packages",
                                        affected_component=lock_file,
                                        cve_id=None,
                                        recommendation="Update dependencies and regenerate lock file",
                                        cvss_score=6.0,
                                        discovered_at=current_time,
                                    )
                                )
                    except Exception:
                        continue

        except Exception:
            pass

        return vulnerabilities

    async def assess_compliance(self, standard: str) -> ComplianceStatus:
        """Assess compliance with specific security standard"""
        findings = []
        current_time = datetime.now()

        if standard.lower() == "pci_dss":
            # PCI DSS assessment
            score = 100.0

            # Requirement 3: Protect cardholder data
            if not os.getenv("ENCRYPTION_KEY"):
                findings.append("Missing encryption for cardholder data")
                score -= 20

            # Requirement 4: Strong access control
            if os.getenv("RBAC_ENABLED", "true").lower() != "true":
                findings.append("Insufficient access control measures")
                score -= 15

            # Requirement 6: Regular testing
            if not self._has_security_testing():
                findings.append("No evidence of regular security testing")
                score -= 10

            # Requirement 7: Secure network
            if (
                os.getenv("ENVIRONMENT", "development").lower() == "production"
                and os.getenv("DEBUG", "false").lower() == "true"
            ):
                findings.append("Debug mode enabled in production")
                score -= 25

            status = "compliant" if score >= 90 else "partial" if score >= 70 else "non_compliant"

        elif standard.lower() == "gdpr":
            # GDPR assessment
            score = 100.0

            # Data protection
            if not os.getenv("ENCRYPTION_KEY"):
                findings.append("Insufficient data protection measures")
                score -= 20

            # Right to be forgotten
            if not self._has_data_deletion_capability():
                findings.append("No data deletion mechanism for GDPR right to be forgotten")
                score -= 15

            # Consent management
            if not self._has_consent_management():
                findings.append("No consent management system")
                score -= 10

            # Data breach notification
            if not self._has_breach_notification():
                findings.append("No breach notification system")
                score -= 15

            status = "compliant" if score >= 90 else "partial" if score >= 70 else "non_compliant"

        elif standard.lower() == "soc2":
            # SOC 2 Type II assessment
            score = 100.0

            # Security controls
            if not self._has_monitoring():
                findings.append("Insufficient security monitoring")
                score -= 20

            # Incident response
            if not self._has_incident_response():
                findings.append("No incident response procedure")
                score -= 15

            # Change management
            if not self._has_change_management():
                findings.append("No formal change management process")
                score -= 10

            # Data classification
            if not self._has_data_classification():
                findings.append("No data classification system")
                score -= 10

            status = "compliant" if score >= 90 else "partial" if score >= 70 else "non_compliant"

        else:
            return ComplianceStatus(
                standard=standard,
                status="not_assessed",
                score=0.0,
                findings=["Standard not supported for assessment"],
                last_assessed=current_time,
            )

        return ComplianceStatus(
            standard=standard,
            status=status,
            score=score,
            findings=findings,
            last_assessed=current_time,
        )

    def _has_security_testing(self) -> bool:
        """Check if security testing is configured"""
        # Look for security testing configurations
        security_tools = ["safety", "bandit", "semgrep", "snyk"]
        return any(tool in os.getenv("SECURITY_TOOLS", "") for tool in security_tools)

    def _has_data_deletion_capability(self) -> bool:
        """Check if data deletion mechanisms exist"""
        # Look for user data deletion endpoints/services
        return os.getenv("DATA_RETENTION_ENABLED", "true").lower() == "true"

    def _has_consent_management(self) -> bool:
        """Check if consent management is available"""
        return os.getenv("CONSENT_MANAGEMENT", "false").lower() == "true"

    def _has_breach_notification(self) -> bool:
        """Check if breach notification is configured"""
        return os.getenv("BREACH_NOTIFICATION_ENABLED", "false").lower() == "true"

    def _has_monitoring(self) -> bool:
        """Check if security monitoring is enabled"""
        return os.getenv("SECURITY_MONITORING", "true").lower() == "true"

    def _has_incident_response(self) -> bool:
        """Check if incident response is configured"""
        return os.getenv("INCIDENT_RESPONSE", "false").lower() == "true"

    def _has_change_management(self) -> bool:
        """Check if change management process exists"""
        return os.getenv("CHANGE_MANAGEMENT", "false").lower() == "true"

    def _has_data_classification(self) -> bool:
        """Check if data classification system exists"""
        return os.getenv("DATA_CLASSIFICATION", "false").lower() == "true"

    async def generate_comprehensive_security_report(self) -> Dict[str, Any]:
        """Generate comprehensive security assessment report"""
        current_time = datetime.now()

        # Run all security scans
        scan_tasks = [
            self.scan_authentication_security(),
            self.scan_authorization_security(),
            self.scan_input_validation_security(),
            self.scan_configuration_security(),
            self.scan_dependency_vulnerabilities(),
        ]

        results = await asyncio.gather(*scan_tasks, return_exceptions=True)

        # Collect all vulnerabilities
        all_vulnerabilities = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                # Add scan error as vulnerability
                all_vulnerabilities.append(
                    SecurityVulnerability(
                        severity="medium",
                        category="scan_error",
                        title=f"Security Scan Error {i}",
                        description=f"Security scanning module {i} failed: {str(result)}",
                        affected_component="security_scanner",
                        cve_id=None,
                        recommendation="Fix scanning configuration and retry",
                        cvss_score=5.0,
                        discovered_at=current_time,
                    )
                )
            else:
                all_vulnerabilities.extend(result)

        # Assess compliance for all standards
        compliance_standards = ["pci_dss", "gdpr", "soc2"]
        compliance_results = {}

        for standard in compliance_standards:
            compliance_results[standard] = await self.assess_compliance(standard)

        # Calculate overall security score
        critical_count = sum(1 for v in all_vulnerabilities if v.severity == "critical")
        high_count = sum(1 for v in all_vulnerabilities if v.severity == "high")
        medium_count = sum(1 for v in all_vulnerabilities if v.severity == "medium")
        low_count = sum(1 for v in all_vulnerabilities if v.severity == "low")

        # Weighted scoring
        base_score = 100.0
        base_score -= critical_count * 25
        base_score -= high_count * 15
        base_score -= medium_count * 8
        base_score -= low_count * 3

        security_score = max(0, min(100, base_score))

        # Determine overall risk level
        if critical_count > 0:
            risk_level = "critical"
        elif high_count > 2:
            risk_level = "high"
        elif high_count > 0 or medium_count > 3:
            risk_level = "medium"
        elif medium_count > 0 or low_count > 5:
            risk_level = "low"
        else:
            risk_level = "minimal"

        # Generate recommendations
        recommendations = []

        # Priority recommendations based on findings
        if critical_count > 0:
            recommendations.append("CRITICAL: Address all critical vulnerabilities immediately")

        if high_count > 0:
            recommendations.append("HIGH: Prioritize high-severity vulnerabilities within 30 days")

        if any("authentication" in v.category for v in all_vulnerabilities):
            recommendations.append("Strengthen authentication mechanisms with MFA")

        if any("authorization" in v.category for v in all_vulnerabilities):
            recommendations.append("Review and tighten access controls")

        if any("injection" in v.category for v in all_vulnerabilities):
            recommendations.append("Implement comprehensive input validation and parameterized queries")

        if any("config" in v.category for v in all_vulnerabilities):
            recommendations.append("Secure configuration files and environment variables")

        if any("dependency" in v.category for v in all_vulnerabilities):
            recommendations.append("Update all vulnerable dependencies immediately")

        report = {
            "security_score": security_score,
            "risk_level": risk_level,
            "scan_timestamp": current_time.isoformat(),
            "summary": {
                "total_vulnerabilities": len(all_vulnerabilities),
                "critical": critical_count,
                "high": high_count,
                "medium": medium_count,
                "low": low_count,
                "scan_modules_executed": len(scan_tasks),
                "scan_errors": sum(1 for r in results if isinstance(r, Exception)),
            },
            "vulnerabilities_by_category": {
                "authentication": [v.to_dict() for v in all_vulnerabilities if v.category == "authentication"],
                "authorization": [v.to_dict() for v in all_vulnerabilities if v.category == "authorization"],
                "injection": [v.to_dict() for v in all_vulnerabilities if v.category == "injection"],
                "xss": [v.to_dict() for v in all_vulnerabilities if v.category == "xss"],
                "configuration": [v.to_dict() for v in all_vulnerabilities if v.category == "config"],
                "dependency": [v.to_dict() for v in all_vulnerabilities if v.category == "dependency"],
                "network": [v.to_dict() for v in all_vulnerabilities if v.category == "network"],
                "scan_error": [v.to_dict() for v in all_vulnerabilities if v.category == "scan_error"],
            },
            "compliance_assessment": {standard: comp.to_dict() for standard, comp in compliance_results.items()},
            "recommendations": recommendations,
            "next_steps": [
                "1. Address critical and high vulnerabilities immediately",
                "2. Implement automated security scanning in CI/CD pipeline",
                "3. Schedule regular penetration testing",
                "4. Establish security incident response procedures",
                "5. Monitor for new vulnerabilities regularly",
            ],
        }

        return report


# Global security scanner instance
security_scanner = AdvancedSecurityScanner()