Implementado un wireShark

He creado en un documento temporal un mini wireshark
para analizar los paquetes en red.

pruebas.py

@@ -1,32 +1,77 @@
-import socket 
+from scapy.all import sniff
-import ipaddress 
+from scapy.layers.inet import IP, TCP, UDP, ICMP
-import nmap 
+from scapy.packet import Raw
-import time
+import os
-import threading
+import datetime
 
-def get_network_range(): 
+def save_packet_to_file(packet):
-    # Obtener la IP local 
+    """Guarda el paquete en un archivo de registro para análisis posterior."""
-    hostname = socket.gethostname() 
+    with open("packet_log.txt", "a") as log_file:
-    local_ip = socket.gethostbyname(hostname) 
+        log_file.write(f"{datetime.datetime.now()} - {packet.summary()}\n")
-    # Suponemos una máscara de red /24 
+        log_file.write(f"{packet.show(dump=True)}\n\n")
-    network = ipaddress.ip_network(f"{local_ip}/24", strict=False) 
-    return str(network) 
 
-def scan_network(ip_range):
+def packet_callback(packet):
-    # Crear un escáner Nmap 
+    """Procesa y muestra información del paquete capturado."""
-    nm = nmap.PortScanner() 
+    if IP in packet:
-    print(f"Escaneando la red {ip_range}...") 
+        ip_layer = packet[IP]
-    nm.scan(hosts=ip_range, arguments='-sn') 
+        print(f"\n[+] Capturado un paquete:")
-    devices = [] 
+        print(f"    Origen: {ip_layer.src}")
-    for host in nm.all_hosts(): 
+        print(f"    Destino: {ip_layer.dst}")
-        if nm[host].state() == "up": 
+        print(f"    Protocolo: {ip_layer.proto}")
-            mac = nm[host]['addresses'].get('mac', 'No MAC address found') 
-            devices.append({'ip': nm[host]['addresses']['ipv4'], 'mac': mac}) 
-    return devices 
 
+        if TCP in packet:
+            tcp_layer = packet[TCP]
+            print(f"    Puerto origen: {tcp_layer.sport}")
+            print(f"    Puerto destino: {tcp_layer.dport}")
+        elif UDP in packet:
+            udp_layer = packet[UDP]
+            print(f"    Puerto origen: {udp_layer.sport}")
+            print(f"    Puerto destino: {udp_layer.dport}")
+        elif ICMP in packet:
+            print("    Tipo de ICMP detectado.")
 
-network_range = get_network_range() 
+        if Raw in packet:
-devices = scan_network(network_range) 
+            payload = packet[Raw].load.decode(errors="ignore")
-# Mostrar resultados 
+            print(f"    Datos (Raw): {payload}")
-for device in devices: 
+            if "youtube.com" in payload or "YouTube" in payload:
-    print(f"IP: {device['ip']}, MAC: {device['mac']}")
+                print(f"\n[!!!] Posible enlace de YouTube detectado:")
+                print(payload)
+
+        print(f"[Paquete completo]: {packet.summary()}")
+
+        # Guardar el paquete en un archivo
+        save_packet_to_file(packet)
+
+def start_sniffer():
+    """Inicia la captura de paquetes."""
+    print("Iniciando captura de paquetes. Presiona Ctrl+C para detener.")
+    print("Opciones disponibles:")
+    print("1. Capturar todo el tráfico IP")
+    print("2. Filtrar por TCP")
+    print("3. Filtrar por UDP")
+    print("4. Filtrar por ICMP")
+    print("5. Salir")
+
+    choice = input("Selecciona una opción (1-5): ")
+    try:
+        if choice == "1":
+            sniff(prn=packet_callback, filter="ip", store=0)
+        elif choice == "2":
+            sniff(prn=packet_callback, filter="tcp", store=0)
+        elif choice == "3":
+            sniff(prn=packet_callback, filter="udp", store=0)
+        elif choice == "4":
+            sniff(prn=packet_callback, filter="icmp", store=0)
+        elif choice == "5":
+            print("Saliendo del programa.")
+            return
+        else:
+            print("Opción no válida. Intenta de nuevo.")
+            start_sniffer()
+    except KeyboardInterrupt:
+        print("\nCaptura detenida.")
+
+if __name__ == "__main__":
+    if os.path.exists("packet_log.txt"):
+        os.remove("packet_log.txt")
+    start_sniffer()