network_scanner.py - master - marsela13/practic

network_scanner.py
551 строка25 KB
marsela13
Upload files to ""
25 дек 2025, 15:33
25 дек 2025, 15:33b6f708d
"""
Модуль сетевого сканирования.

Содержит класс NetworkScanner для обнаружения активных устройств в сети,
сканирования портов и сбора информации о хостах.

Основные возможности:
- Ping-сканирование диапазонов IP-адресов
- TCP-сканирование портов с поддержкой многопоточности
- Определение имен хостов (hostname)
- Экспорт результатов в JSON и CSV форматы
- Прогресс-отслеживание и возможность отмены операции

Зависимости:
- Python 3.8+
- Стандартные библиотеки: socket, subprocess, threading, ipaddress
- Для дополнительных функций: tabulate (опционально)

Примечание: Для сканирования сети требуются соответствующие права доступа.
"""

import socket
import subprocess
import platform
import ipaddress
import threading
import queue
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
import csv
from datetime import datetime
from typing import List, Dict, Tuple, Optional
import re


class NetworkScanner:
    """
    Основной класс для сканирования сети.
    
    Предоставляет функционал для обнаружения активных устройств,
    сканирования портов и сбора информации о сетевых узлах.
    
    Атрибуты:
        COMMON_PORTS (List[int]): Список наиболее часто используемых портов
        max_threads (int): Максимальное количество параллельных потоков
        timeout (float): Таймаут сетевых операций в секундах
        results (List[Dict]): Результаты последнего сканирования
        progress_callback (Callable): Callback-функция для отслеживания прогресса
        is_cancelled (bool): Флаг отмены текущей операции сканирования
    """
    
    # Наиболее часто используемые TCP порты для сканирования
    COMMON_PORTS = [21, 22, 23, 25, 53, 80, 110, 111, 135, 139, 143, 443, 
                    445, 993, 995, 1723, 3306, 3389, 5900, 8080]
    
    def __init__(self, max_threads: int = 50, timeout: float = 1.0):
        """
        Инициализация объекта NetworkScanner.
        
        Args:
            max_threads: Максимальное количество потоков для параллельного сканирования
                         Рекомендуется: 10-100 в зависимости от пропускной способности сети
            timeout: Таймаут для сетевых операций в секундах
                    Рекомендуется: 0.5-2 секунды в зависимости от скорости сети
        
        Пример:
            >>> scanner = NetworkScanner(max_threads=50, timeout=1.0)
        """
        self.max_threads = max_threads
        self.timeout = timeout
        self.results = []  # Список для хранения результатов сканирования
        self.progress_callback = None  # Callback для обновления прогресса
        self.is_cancelled = False  # Флаг для отмены операции
        
    def set_progress_callback(self, callback):
        """
        Установка callback-функции для отслеживания прогресса сканирования.
        
        Args:
            callback: Функция, принимающая параметры:
                     - current (int): Текущий обработанный хост
                     - total (int): Общее количество хостов
                     - found (int): Найденных активных хостов
                     - result (Dict): Информация о последнем найденном хосте
        
        Пример:
            >>> def progress_update(current, total, found, result):
            ...     print(f"Прогресс: {current}/{total}, найдено: {found}")
            >>> scanner.set_progress_callback(progress_update)
        """
        self.progress_callback = callback
        
    def cancel_scan(self):
        """
        Отмена текущей операции сканирования.
        
        Устанавливает флаг is_cancelled в True, что приводит к
        постепенному прекращению операций сканирования.
        """
        self.is_cancelled = True
        
    def parse_ip_range(self, ip_range: str) -> List[str]:
        """
        Парсинг строки с диапазоном IP-адресов.
        
        Поддерживает следующие форматы:
        1. CIDR нотация: 192.168.1.0/24
        2. Диапазон с полными IP: 192.168.1.10-192.168.1.50
        3. Диапазон с последним октетом: 192.168.1.1-100
        4. Одиночный IP: 192.168.1.1
        
        Args:
            ip_range: Строка с диапазоном IP-адресов
        
        Returns:
            List[str]: Список IP-адресов для сканирования
        
        Raises:
            ValueError: При некорректном формате входных данных
        
        Пример:
            >>> scanner.parse_ip_range("192.168.1.0/24")
            ['192.168.1.1', '192.168.1.2', ..., '192.168.1.254']
        """
        ip_list = []
        
        try:
            # Проверка формата CIDR
            if '/' in ip_range:
                network = ipaddress.ip_network(ip_range, strict=False)
                ip_list = [str(ip) for ip in network.hosts()]
                
            # Проверка формата диапазона
            elif '-' in ip_range:
                if '.' in ip_range.split('-')[1]:  # Полный диапазон
                    start_ip, end_ip = ip_range.split('-')
                    start = ipaddress.ip_address(start_ip.strip())
                    end = ipaddress.ip_address(end_ip.strip())
                    
                    current = int(start)
                    end_int = int(end)
                    
                    while current <= end_int:
                        ip_list.append(str(ipaddress.ip_address(current)))
                        current += 1
                else:  # Сокращенный диапазон (192.168.1.1-100)
                    base_ip = ip_range.split('-')[0]
                    base_parts = base_ip.split('.')
                    if len(base_parts) == 4:
                        start_num = int(ip_range.split('-')[1])
                        for i in range(1, start_num + 1):
                            ip_list.append(f"{'.'.join(base_parts[:3])}.{i}")
                            
            # Одиночный IP
            else:
                ip_list.append(ip_range)
                
        except Exception as e:
            error_msg = f"Ошибка парсинга IP диапазона '{ip_range}': {e}"
            print(error_msg)
            raise ValueError(error_msg)
            
        return ip_list
    
    def ping_host(self, ip: str) -> bool:
        """
        Проверка доступности хоста с помощью ICMP ping.
        
        Использует системную команду ping с параметрами,
        специфичными для каждой операционной системы.
        
        Args:
            ip: IP-адрес хоста для проверки
        
        Returns:
            bool: True если хост отвечает на ping, иначе False
        
        Пример:
            >>> scanner.ping_host("192.168.1.1")
            True
        """
        if self.is_cancelled:
            return False
            
        # Определение параметров ping для разных ОС
        param = '-n' if platform.system().lower() == 'windows' else '-c'
        command = ['ping', param, '1', '-w', str(int(self.timeout * 1000)), ip]
        
        try:
            # Запуск системной команды ping с таймаутом
            result = subprocess.run(command, 
                                   stdout=subprocess.PIPE, 
                                   stderr=subprocess.PIPE,
                                   timeout=self.timeout + 1)
            return result.returncode == 0
        except (subprocess.TimeoutExpired, FileNotFoundError, Exception):
            # Обработка ошибок: таймаут, отсутствие команды ping, другие ошибки
            return False
    
    def get_hostname(self, ip: str) -> str:
        """
        Получение имени хоста по IP-адресу с помощью обратного DNS-запроса.
        
        Args:
            ip: IP-адрес для разрешения в hostname
        
        Returns:
            str: Имя хоста или пустая строка при ошибке
        
        Пример:
            >>> scanner.get_hostname("8.8.8.8")
            'dns.google'
        """
        if self.is_cancelled:
            return ""
            
        try:
            # Выполнение обратного DNS-запроса
            hostname, _, _ = socket.gethostbyaddr(ip)
            return hostname
        except (socket.herror, socket.gaierror, socket.timeout):
            # Ошибки DNS разрешения
            return ""
    
    def scan_port(self, ip: str, port: int) -> bool:
        """
        Проверка доступности TCP порта на указанном хосте.
        
        Метод использует socket.connect_ex() для проверки подключения
        к указанному порту с установленным таймаутом.
        
        Args:
            ip: IP-адрес хоста
            port: Номер TCP порта для проверки (1-65535)
        
        Returns:
            bool: True если порт открыт и принимает соединения, иначе False
        
        Пример:
            >>> scanner.scan_port("192.168.1.1", 80)
            True
        """
        if self.is_cancelled:
            return False
            
        try:
            # Создание TCP сокета
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            
            # Попытка подключения
            result = sock.connect_ex((ip, port))
            sock.close()
            
            # Код 0 означает успешное подключение
            return result == 0
        except (socket.timeout, socket.error, OSError):
            # Ошибки сети или сокета
            return False
    
    def scan_ports(self, ip: str, ports: List[int]) -> List[int]:
        """
        Многопоточное сканирование списка портов на указанном хосте.
        
        Args:
            ip: IP-адрес хоста для сканирования
            ports: Список номеров портов для проверки
        
        Returns:
            List[int]: Список открытых портов
        
        Пример:
            >>> scanner.scan_ports("192.168.1.1", [22, 80, 443])
            [80, 443]
        """
        if self.is_cancelled:
            return []
            
        open_ports = []
        
        # Использование ThreadPoolExecutor для параллельного сканирования портов
        with ThreadPoolExecutor(max_workers=min(self.max_threads, len(ports))) as executor:
            # Создание задач для сканирования каждого порта
            future_to_port = {executor.submit(self.scan_port, ip, port): port for port in ports}
            
            # Обработка завершенных задач
            for future in as_completed(future_to_port):
                if self.is_cancelled:
                    break
                    
                port = future_to_port[future]
                try:
                    if future.result():
                        open_ports.append(port)
                except Exception:
                    # Игнорирование ошибок в отдельных задачах
                    pass
                    
        return sorted(open_ports)
    
    def scan_single_host(self, ip: str, ports: List[int]) -> Optional[Dict]:
        """
        Полное сканирование одного хоста.
        
        Выполняет последовательно:
        1. Проверку доступности (ping)
        2. Получение hostname (обратный DNS)
        3. Сканирование указанных портов
        
        Args:
            ip: IP-адрес хоста
            ports: Список портов для сканирования
        
        Returns:
            Optional[Dict]: Словарь с информацией о хосте или None если хост недоступен
        
        Структура возвращаемого словаря:
            {
                'ip': str,           # IP-адрес
                'hostname': str,     # Имя хоста или пустая строка
                'status': str,       # 'Online' или 'Offline'
                'open_ports': List[int],  # Список открытых портов
                'timestamp': str     # Время обнаружения в ISO формате
            }
        """
        if self.is_cancelled:
            return None
            
        # Шаг 1: Проверка доступности хоста
        is_alive = self.ping_host(ip)
        
        if not is_alive:
            return None
            
        # Шаг 2: Получение имени хоста
        hostname = self.get_hostname(ip)
        
        # Шаг 3: Сканирование портов (если указаны)
        open_ports = self.scan_ports(ip, ports) if ports else []
        
        return {
            'ip': ip,
            'hostname': hostname,
            'status': 'Online',
            'open_ports': open_ports,
            'timestamp': datetime.now().isoformat()
        }
    
    def scan_network(self, ip_range: str, ports: List[int] = None) -> List[Dict]:
        """
        Основной метод сканирования сети.
        
        Выполняет многопоточное сканирование указанного диапазона IP-адресов.
        
        Args:
            ip_range: Диапазон IP-адресов для сканирования
            ports: Список портов для сканирования (по умолчанию - COMMON_PORTS)
        
        Returns:
            List[Dict]: Список словарей с информацией о найденных хостах
        
        Исключения:
            ValueError: При некорректном формате ip_range
            RuntimeError: При ошибках во время сканирования
        
        Пример:
            >>> results = scanner.scan_network("192.168.1.1-10", [22, 80])
            >>> len(results)
            3
        """
        self.is_cancelled = False
        self.results = []
        
        if ports is None:
            ports = self.COMMON_PORTS
        
        # Парсинг диапазона IP
        ip_list = self.parse_ip_range(ip_range)
        total_hosts = len(ip_list)
        
        if total_hosts == 0:
            print("Предупреждение: Не получено IP-адресов для сканирования")
            return []
        
        print(f"Начато сканирование {total_hosts} хостов...")
        
        # Многопоточное сканирование хостов
        with ThreadPoolExecutor(max_workers=min(self.max_threads, total_hosts)) as executor:
            # Создание задач для каждого IP-адреса
            future_to_ip = {executor.submit(self.scan_single_host, ip, ports): ip 
                           for ip in ip_list}
            
            completed = 0
            for future in as_completed(future_to_ip):
                if self.is_cancelled:
                    print("Сканирование отменено пользователем")
                    break
                    
                completed += 1
                ip = future_to_ip[future]
                
                try:
                    result = future.result()
                    if result:
                        self.results.append(result)
                        
                        # Вызов callback для обновления прогресса (если установлен)
                        if self.progress_callback:
                            self.progress_callback(completed, total_hosts, 
                                                  len(self.results), result)
                except Exception as e:
                    print(f"Ошибка при сканировании {ip}: {e}")
                
                # Вывод прогресса в консоль
                progress_percent = (completed / total_hosts) * 100
                print(f"Прогресс: {progress_percent:.1f}% ({completed}/{total_hosts})")
        
        print(f"Сканирование завершено. Найдено {len(self.results)} активных хостов.")
        return self.results
    
    def export_to_json(self, filename: str) -> bool:
        """
        Экспорт результатов сканирования в JSON файл.
        
        Args:
            filename: Путь к файлу для сохранения
        
        Returns:
            bool: True если экспорт успешен, иначе False
        
        Пример:
            >>> scanner.export_to_json("results.json")
            True
        """
        try:
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(self.results, f, indent=2, ensure_ascii=False)
            print(f"Результаты успешно экспортированы в {filename}")
            return True
        except (IOError, PermissionError, json.JSONEncodeError) as e:
            print(f"Ошибка экспорта в JSON '{filename}': {e}")
            return False
    
    def export_to_csv(self, filename: str) -> bool:
        """
        Экспорт результатов сканирования в CSV файл.
        
        Создает таблицу, где каждый хост - строка, а каждый порт - столбец.
        
        Args:
            filename: Путь к файлу для сохранения
        
        Returns:
            bool: True если экспорт успешен, иначе False
        
        Пример:
            >>> scanner.export_to_csv("results.csv")
            True
        """
        try:
            with open(filename, 'w', newline='', encoding='utf-8') as f:
                if not self.results:
                    print("Предупреждение: Нет данных для экспорта в CSV")
                    return False
                
                # Сбор всех уникальных портов из результатов
                all_ports = set()
                for host in self.results:
                    all_ports.update(host['open_ports'])
                
                # Создание заголовков столбцов
                fieldnames = ['IP', 'Hostname', 'Status', 'Timestamp']
                fieldnames.extend([f'Port_{p}' for p in sorted(all_ports)])
                
                writer = csv.DictWriter(f, fieldnames=fieldnames)
                writer.writeheader()
                
                # Запись данных по каждому хосту
                for host in self.results:
                    row = {
                        'IP': host['ip'],
                        'Hostname': host['hostname'],
                        'Status': host['status'],
                        'Timestamp': host['timestamp']
                    }
                    
                    # Отметка состояния каждого порта
                    for port in sorted(all_ports):
                        row[f'Port_{port}'] = 'Open' if port in host['open_ports'] else 'Closed'
                    
                    writer.writerow(row)
            
            print(f"Результаты успешно экспортированы в {filename}")
            return True
        except (IOError, PermissionError, csv.Error) as e:
            print(f"Ошибка экспорта в CSV '{filename}': {e}")
            return False
    
    def print_results_table(self):
        """
        Вывод результатов сканирования в виде форматированной таблицы в консоль.
        
        Требует установленной библиотеки tabulate.
        
        Пример:
            >>> scanner.print_results_table()
            +-------------+-------------------+---------+-------------------+
            | IP          | Hostname          | Status  | Открытые порты   |
            +-------------+-------------------+---------+-------------------+
            | 192.168.1.1 | router.local      | Online  | 80, 443          |
            +-------------+-------------------+---------+-------------------+
        """
        try:
            from tabulate import tabulate
        except ImportError:
            print("Для вывода таблицы установите библиотеку: pip install tabulate")
            return
        
        if not self.results:
            print("Активные хосты не обнаружены.")
            return
        
        # Подготовка данных для таблицы
        table_data = []
        for host in self.results:
            ports_str = ', '.join(map(str, host['open_ports'])) if host['open_ports'] else 'Нет'
            table_data.append([
                host['ip'],
                host['hostname'] or 'Не определено',
                host['status'],
                ports_str
            ])
        
        headers = ["IP", "Hostname", "Status", "Открытые порты"]
        print("\n" + tabulate(table_data, headers=headers, tablefmt="grid"))
        
        # Вывод статистики
        print(f"\n{'='*50}")
        print("СТАТИСТИКА СКАНИРОВАНИЯ:")
        print(f"{'='*50}")
        print(f"Всего просканировано хостов: {len(self.results)}")
        
        open_ports_count = sum(len(h['open_ports']) for h in self.results)
        print(f"Всего обнаружено открытых портов: {open_ports_count}")
        
        if self.results:
            avg_ports = open_ports_count / len(self.results)
            print(f"Среднее количество открытых портов на хост: {avg_ports:.1f}")
        
        print(f"{'='*50}")