packetsniffer.py

import scapy.all as scapy
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.l2 import Ether
import argparse
import signal
import sys
import time
from datetime import datetime
import threading
from queue import Queue
import psutil
import logging
from typing import Optional, Dict

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('network_sniffer.log'),
        logging.StreamHandler()
    ]
)

class NetworkSniffer:
    def __init__(self, interface: Optional[str] = None, filter_str: str = "", max_packets: int = 0):
        """Initialize network sniffer with given parameters."""
        self.interface = interface or self._get_default_interface()
        self.filter_str = filter_str
        self.max_packets = max_packets
        self.packet_count = 0
        self.packet_queue = Queue()
        self.running = False
        self.protocol_stats: Dict[str, int] = {}
        
    def _get_default_interface(self) -> str:
        """Get default network interface."""
        try:
            interfaces = psutil.net_if_addrs()
            for iface in interfaces:
                if iface != 'lo' and '127.0.0.1' not in str(interfaces[iface]):
                    return iface
            raise ValueError("No suitable network interface found")
        except Exception as e:
            logging.error(f"Error getting default interface: {e}")
            sys.exit(1)

    def _handle_packet(self, packet: scapy.Packet) -> None:
        """Process captured packet and extract relevant information."""
        self.packet_count += 1
        self.packet_queue.put(packet)
        
        # Update protocol statistics
        protocol = packet.getlayer(1).name if packet.haslayer(IP) else 'Unknown'
        self.protocol_stats[protocol] = self.protocol_stats.get(protocol, 0) + 1
        
        # Detailed packet analysis
        try:
            self._analyze_packet(packet)
        except Exception as e:
            logging.error(f"Error analyzing packet: {e}")

    def _analyze_packet(self, packet: scapy.Packet) -> None:
        """Analyze packet structure and content."""
        timestamp = datetime.fromtimestamp(packet.time).strftime('%Y-%m-%d %H:%M:%S.%f')
        
        # Ethernet layer
        if packet.haslayer(Ether):
            eth = packet[Ether]
            src_mac = eth.src
            dst_mac = eth.dst
        else:
            src_mac = dst_mac = "N/A"

        # IP layer
        if packet.haslayer(IP):
            ip = packet[IP]
            src_ip = ip.src
            dst_ip = ip.dst
            protocol = ip.proto
            ttl = ip.ttl
        else:
            src_ip = dst_ip = protocol = ttl = "N/A"

        # Transport layer
        payload = port_info = "N/A"
        if packet.haslayer(TCP):
            tcp = packet[TCP]
            port_info = f"{tcp.sport} -> {tcp.dport}"
            payload = str(tcp.payload)[:50] if tcp.payload else "Empty"
            protocol = "TCP"
        elif packet.haslayer(UDP):
            udp = packet[UDP]
            port_info = f"{udp.sport} -> {udp.dport}"
            payload = str(udp.payload)[:50] if udp.payload else "Empty"
            protocol = "UDP"
        elif packet.haslayer(ICMP):
            icmp = packet[ICMP]
            protocol = "ICMP"
            payload = str(icmp.payload)[:50] if icmp.payload else "Empty"

        # Log packet details
        log_message = (
            f"\n[Packet #{self.packet_count}] {timestamp}\n"
            f"MAC: {src_mac} -> {dst_mac}\n"
            f"IP: {src_ip} -> {dst_ip} | Protocol: {protocol} | TTL: {ttl}\n"
            f"Ports: {port_info}\n"
            f"Payload: {payload}..."
        )
        logging.info(log_message)

    def _display_stats(self) -> None:
        """Periodically display capture statistics."""
        while self.running:
            time.sleep(10)
            if self.packet_count > 0:
                stats = (
                    f"\n=== Capture Statistics ===\n"
                    f"Total Packets: {self.packet_count}\n"
                    f"Protocols: {self.protocol_stats}\n"
                    f"Queue Size: {self.packet_queue.qsize()}\n"
                    f"=========================\n"
                )
                logging.info(stats)

    def start(self) -> None:
        """Start packet capture."""
        self.running = True
        logging.info(f"Starting packet capture on interface {self.interface}")
        logging.info(f"Filter: {self.filter_str if self.filter_str else 'None'}")
        
        # Start stats display thread
        stats_thread = threading.Thread(target=self._display_stats, daemon=True)
        stats_thread.start()

        try:
            scapy.sniff(
                iface=self.interface,
                filter=self.filter_str,
                prn=self._handle_packet,
                count=self.max_packets,
                store=False
            )
        except Exception as e:
            logging.error(f"Error during packet capture: {e}")
        finally:
            self.stop()

    def stop(self) -> None:
        """Stop packet capture."""
        self.running = False
        logging.info(f"\nCapture stopped. Total packets captured: {self.packet_count}")
        logging.info(f"Final protocol statistics: {self.protocol_stats}")

def signal_handler(sig, frame):
    """Handle Ctrl+C gracefully."""
    logging.info("\nInterrupt received, stopping sniffer...")
    sys.exit(0)

def main():
    """Main function to parse arguments and run sniffer."""
    parser = argparse.ArgumentParser(description="Advanced Network Packet Sniffer")
    parser.add_argument("-i", "--interface", help="Network interface to sniff on")
    parser.add_argument("-f", "--filter", default="", help="BPF filter string (e.g., 'tcp port 80')")
    parser.add_argument("-c", "--count", type=int, default=0, help="Number of packets to capture (0 for unlimited)")
    
    args = parser.parse_args()

    # Set up signal handler
    signal.signal(signal.SIGINT, signal_handler)

    # Initialize and start sniffer
    sniffer = NetworkSniffer(args.interface, args.filter, args.count)
    sniffer.start()

if __name__ == "__main__":
    main()