scanner/vulnerability_scanner.py - master - manokuzCS/Web_Scanner_SQL_XSS_CSRF

scanner/vulnerability_scanner.py
197 строк8 KB
Manokuz
Fixes and add a CLI mode
28 окт 2025, 12:33
28 окт 2025, 12:3358fec7d
import asyncio
from typing import Dict, List, Any, Set, Optional
from datetime import datetime
import time
from urllib.parse import urlparse

from PyQt5.QtCore import QObject, pyqtSignal

from utils.logger import log_and_notify
from .html_processor import HTMLProcessor
from .network_manager import NetworkManager

class VulnerabilityScanner(QObject):
    """Сканер уязвимостей"""
    
    # Сигналы
    scan_started = pyqtSignal(str)
    scan_finished = pyqtSignal(str)
    error_occurred = pyqtSignal(str)
    vulnerability_found = pyqtSignal(str, str, str)
    
    def __init__(self, scan_options: Optional[Dict[str, Any]] = None):
        super().__init__()
        self._initialize_state()
        self._setup_options(scan_options or {})
        self.html_processor = HTMLProcessor()
        
    def _initialize_state(self) -> None:
        """Инициализация состояния сканера"""
        self.vulnerabilities: Dict[str, List[Dict[str, Any]]] = {
            'sql': [], 'xss': [], 'csrf': []
        }
        self.scanned_urls: Set[str] = set()
        self.scanned_forms: Set[str] = set()
        self.unscanned_urls: Set[str] = set()
        self.start_time = time.time()
        self.base_url = ""
        self.should_stop = False
        self._is_paused = False
    
    def _setup_options(self, options: Dict[str, Any]) -> None:
        """Настройка параметров сканирования"""
        self.max_depth = options.get('max_depth', 3)
        self.timeout = options.get('timeout', 30)
        self.max_retries = options.get('max_retries', 3)
        self.concurrent_scans = options.get('concurrent_scans', 5)
        self.scan_types = options.get('scan_types', ['sql', 'xss', 'csrf'])
    
    async def scan(self, url: str) -> Dict[str, Any]:
        """Запуск сканирования"""
        try:
            self.base_url = url
            self.start_time = time.time()
            base_domain = urlparse(url).netloc
            
            async with NetworkManager(
                timeout=self.timeout,
                max_retries=self.max_retries,
                max_concurrent=self.concurrent_scans
            ) as network:
                await self._scan_site(network, base_domain)
            
            scan_duration = time.time() - self.start_time
            return self._prepare_results(scan_duration)
            
        except Exception as e:
            log_and_notify('error', f"Error in scan method: {e}")
            return self._prepare_error_results(str(e))
    
    async def _scan_site(self, network: NetworkManager, base_domain: str) -> None:
        """Сканирование сайта"""
        try:
            response = await network.request('GET', self.base_url)
            if not response:
                return
            
            _, html = response
            links, forms = self.html_processor.extract_links_and_forms(html, self.base_url)
            
            # Фильтруем ссылки того же домена
            same_domain_links = {
                link for link in links 
                if self.html_processor.is_same_domain(link, base_domain)
            }
            
            # Создаем задачи для сканирования
            tasks: List[asyncio.Task[None]] = []
            for link in same_domain_links:
                if link not in self.scanned_urls:
                    tasks.append(asyncio.create_task(self._scan_url(network, link)))
            
            for form in forms:
                form_hash = self.html_processor.get_form_hash(form)
                if form_hash not in self.scanned_forms:
                    tasks.append(asyncio.create_task(self._scan_form(network, form, self.base_url)))
            
            await asyncio.gather(*tasks)
            
        except Exception as e:
            log_and_notify('error', f"Error scanning site: {e}")
    
    async def _scan_url(self, network: NetworkManager, url: str) -> None:
        """Сканирование отдельного URL"""
        if url in self.scanned_urls or self.should_stop:
            return
        
        try:
            self.scanned_urls.add(url)
            response = await network.request('GET', url)
            if not response:
                return
            
            _, html = response
            await self._scan_content(url, html)
            
        except Exception as e:
            log_and_notify('error', f"Error scanning URL {url}: {e}")
            self.unscanned_urls.add(url)
    
    async def _scan_form(self, network: NetworkManager, form: Any, base_url: str) -> None:
        """Сканирование формы"""
        form_hash = self.html_processor.get_form_hash(form)
        if form_hash in self.scanned_forms or self.should_stop:
            return
        
        try:
            self.scanned_forms.add(form_hash)
            # Здесь должна быть логика сканирования формы
            # Зависит от конкретных требований к проверке форм
            
        except Exception as e:
            log_and_notify('error', f"Error scanning form: {e}")
    
    async def _scan_content(self, url: str, content: str) -> None:
        """Сканирование содержимого на уязвимости"""
        if 'sql' in self.scan_types:
            await self._scan_sql_injection(url, content)
        if 'xss' in self.scan_types:
            await self._scan_xss(url, content)
        if 'csrf' in self.scan_types:
            await self._scan_csrf(url, content)
    
    async def _scan_sql_injection(self, url: str, content: str) -> None:
        """Сканирование на SQL-инъекции"""
        # Реализация сканирования SQL-инъекций
        pass
    
    async def _scan_xss(self, url: str, content: str) -> None:
        """Сканирование на XSS"""
        # Реализация сканирования XSS
        pass
    
    async def _scan_csrf(self, url: str, content: str) -> None:
        """Сканирование на CSRF"""
        # Реализация сканирования CSRF
        pass
    
    def _prepare_results(self, scan_duration: float) -> Dict[str, Any]:
        """Подготовка результатов сканирования"""
        return {
            'url': self.base_url,
            'scan_types': self.scan_types,
            'duration': scan_duration,
            'total_urls_scanned': len(self.scanned_urls),
            'total_forms_scanned': len(self.scanned_forms),
            'vulnerabilities': self.vulnerabilities,
            'timestamp': datetime.now().isoformat(),
            'total_urls_discovered': len(self.scanned_urls) + len(self.unscanned_urls),
            'unscanned_urls': list(self.unscanned_urls)
        }
    
    def _prepare_error_results(self, error: str) -> Dict[str, Any]:
        """Подготовка результатов в случае ошибки"""
        return {
            'url': self.base_url,
            'scan_types': self.scan_types,
            'duration': 0,
            'total_urls_scanned': 0,
            'total_forms_scanned': 0,
            'vulnerabilities': {'sql': [], 'xss': [], 'csrf': []},
            'timestamp': datetime.now().isoformat(),
            'error': error,
            'total_urls_discovered': 0,
            'unscanned_urls': list(self.unscanned_urls)
        }
    
    def stop(self) -> None:
        """Остановка сканирования"""
        self.should_stop = True
    
    def pause(self) -> None:
        """Приостановка сканирования"""
        self._is_paused = True
    
    def resume(self) -> None:
        """Возобновление сканирования"""
        self._is_paused = False