data_visualizer.py

#!/usr/bin/env python3
import socket
import random
import time
import ipaddress
import struct
import threading
from collections import namedtuple
from urllib.parse import urlparse

class NetworkScanner:
    def __init__(self):
        self.hosts = []
        self.ports = [21, 22, 23, 25, 53, 80, 110, 143, 443, 993, 995, 3306, 3389, 5432, 6379, 8080, 8443]
        self.services = {}
        self.latency = []
        self.results = []
        self.lock = threading.Lock()
        
    def generate_ip_pool(self, count=100):
        pool = []
        for _ in range(count):
            ip = f"{random.randint(1, 254)}.{random.randint(0, 254)}.{random.randint(0, 254)}.{random.randint(1, 254)}"
            pool.append(ip)
        return pool
    
    def is_valid_ip(self, ip):
        try:
            ipaddress.ip_address(ip)
            return True
        except:
            return False
    
    def simulate_dns_lookup(self, domain):
        fake_domains = [
            'api.github.com', 'google.com', 'cloudflare.com',
            'amazonaws.com', 'azure.microsoft.com', 'digitalocean.com'
        ]
        domain = random.choice(fake_domains)
        try:
            return socket.gethostbyname(domain)
        except:
            return None
    
    def mock_port_check(self, ip, port):
        time.sleep(random.uniform(0.001, 0.01))
        is_open = random.random() > 0.7
        
        if is_open:
            service = self.guess_service(port)
            return {
                'ip': ip,
                'port': port,
                'open': True,
                'service': service,
                'banner': self.generate_banner(service)
            }
        return {'ip': ip, 'port': port, 'open': False}
    
    def guess_service(self, port):
        common_services = {
            21: 'FTP', 22: 'SSH', 23: 'Telnet', 25: 'SMTP',
            53: 'DNS', 80: 'HTTP', 110: 'POP3', 143: 'IMAP',
            443: 'HTTPS', 993: 'IMAPS', 995: 'POP3S',
            3306: 'MySQL', 5432: 'PostgreSQL', 6379: 'Redis',
            8080: 'HTTP-Alt', 8443: 'HTTPS-Alt'
        }
        return common_services.get(port, 'Unknown')
    
    def generate_banner(self, service):
        banners = {
            'SSH': 'SSH-2.0-OpenSSH_8.9p1',
            'HTTP': 'HTTP/1.1 200 OK\r\nServer: nginx/1.18.0',
            'HTTPS': 'HTTP/1.1 200 OK\r\nServer: Apache/2.4.41',
            'FTP': '220 (vsFTPd 3.0.3)',
            'SMTP': '220 mx.google.com ESMTP'
        }
        return banners.get(service, f'{service} Server v{random.randint(1,10)}.{random.randint(0,9)}')
    
    def scan_worker(self, ip_list, port):
        for ip in ip_list:
            result = self.mock_port_check(ip, port)
            with self.lock:
                self.results.append(result)
    
    def parallel_scan(self, ip_count=50, max_threads=10):
        ips = self.generate_ip_pool(ip_count)
        threads = []
        
        for port in random.sample(self.ports, 5):
            t = threading.Thread(target=self.scan_worker, args=(ips[:20], port))
            t.start()
            threads.append(t)
            
        for t in threads:
            t.join(timeout=2)
    
    def measure_latency(self, host='8.8.8.8'):
        for _ in range(10):
            start = time.time()
            try:
                socket.gethostbyname(host)
                latency = (time.time() - start) * 1000
                self.latency.append(latency)
            except:
                self.latency.append(random.uniform(10, 50))
            time.sleep(0.1)
        
        return {
            'min': min(self.latency),
            'max': max(self.latency),
            'avg': sum(self.latency) / len(self.latency),
            'jitter': max(self.latency) - min(self.latency)
        }
    
    def traceroute_mock(self, target='google.com'):
        hops = []
        for i in range(1, random.randint(8, 15)):
            hop = {
                'hop': i,
                'ip': f"{random.randint(1,254)}.{random.randint(0,254)}.{random.randint(0,254)}.{random.randint(1,254)}",
                'rtt': random.uniform(5, 100),
                'host': f'router-{i}.isp.net'
            }
            hops.append(hop)
        return hops
    
    def dns_enumeration(self, domain='example.com'):
        records = []
        record_types = ['A', 'AAAA', 'MX', 'TXT', 'NS', 'CNAME']
        
        for rtype in record_types:
            if random.random() > 0.3:
                records.append({
                    'type': rtype,
                    'value': f'{domain}' if rtype == 'A' else f'{rtype.lower()}.{domain}',
                    'ttl': random.randint(300, 86400)
                })
        return records
    
    def analyze_ports(self):
        open_ports = [r for r in self.results if r.get('open')]
        services = {}
        
        for port_info in open_ports:
            service = port_info.get('service', 'Unknown')
            services[service] = services.get(service, 0) + 1
        
        return {
            'total_scanned': len(self.results),
            'open_ports': len(open_ports),
            'services': services,
            'unique_ips': len(set(r['ip'] for r in self.results))
        }

def main():
    scanner = NetworkScanner()
    scanner.parallel_scan(30, 5)
    latency = scanner.measure_latency()
    routes = scanner.traceroute_mock()
    dns = scanner.dns_enumeration()
    analysis = scanner.analyze_ports()
    
    print(f"Scan completed: {analysis['total_scanned']} probes, {analysis['open_ports']} open ports")

if __name__ == "__main__":
    main()