#!/usr/bin/env python3 import socket, random, string, time, os from openpyxl import Workbook from openpyxl.utils import get_column_letter import sys import logging # ====== CONFIGURACIÓN ====== HOST = "127.0.0.1" PUERTO_RANGOS = [(6019, 6061), (7001, 7100), (9001, 9061)] #PUERTO_RANGOS = [(9097, 9097)] REINTENTOS_MAX = 3 TIMEOUT = 3 OUTPUT_FILE = "resultado_test_paquetes.xlsx" TIMEOUT_SACK = 10 # Nuevo tiempo de espera específico para SACK en segundos TIMEOUT_CONNECTION = 5 # Tiempo de espera para la conexión inicial # ====== GENERADOR DE MENSAJES ====== def generar_mensaje(): equipo = ''.join(random.choices(string.digits, k=15)) lon = round(random.uniform(-180,180),6) lat = round(random.uniform(-90,90),6) fh = time.strftime("%Y%m%d%H%M%S") hx = ''.join(random.choices(string.hexdigits.lower(), k=4)) u4 = ''.join(random.choices(string.hexdigits.lower(), k=4)) msg = (f"+RESP:GTFRI,500101,{equipo},GV55W,12703,10,1,1,0.0,229,36.3," f"{lon},{lat},{fh},{hx},0007,1168,16C073,00,0.0,,,,96,110000,,,," f"{fh},{u4}$") return msg.encode(), u4 # devolvemos msg y numero de mensaje # ====== ENVÍO TCP ====== def enviar_tcp(puerto, mensaje, numero): """ Envía un mensaje TCP y espera una respuesta SACK por hasta 10 segundos. """ fecha_envio = time.strftime("%d/%m/%Y %H:%M:%S") for intento in range(REINTENTOS_MAX): sock = None try: # 1. Crear conexión con un tiempo de espera inicial sock = socket.create_connection((HOST, puerto), timeout=TIMEOUT_CONNECTION) # 2. Enviar el mensaje t0 = time.time() sock.sendall(mensaje) # 3. Esperar la respuesta SACK con un tiempo de espera de 10 segundos sock.settimeout(TIMEOUT_SACK) data = sock.recv(1024).decode(errors="ignore") # 4. Si la respuesta contiene SACK, devolver éxito if "SACK" in data: fecha_recepcion = time.strftime("%d/%m/%Y %H:%M:%S") delta = round(time.time() - t0, 2) return ("TCP", puerto, mensaje.decode(), numero, fecha_envio, fecha_recepcion, data.strip(), delta) except socket.timeout: # Capturar específicamente la excepción de tiempo de espera # No se debe reintentar, ya que el SACK no llegó en el tiempo definido logging.error(f"Error de tiempo de espera para SACK. No se recibió respuesta en {TIMEOUT_SACK}s.") return ("TCP", puerto, mensaje.decode(), numero, fecha_envio, "-", "TIMEOUT", -1) except Exception as e: # Capturar otras excepciones (ej. error de conexión) y reintentar logging.warning(f"Error al enviar/recibir paquete TCP (Intento {intento + 1}): {e}") continue finally: # Cerrar el socket en cualquier caso if sock: sock.close() # 5. Si todos los reintentos fallaron (ej. por error de conexión), devolver NO_ACK logging.error("Todos los reintentos de conexión fallaron.") return ("TCP", puerto, mensaje.decode(), numero, fecha_envio, "-", "NO_ACK", -1) # ====== ENVÍO UDP ====== def enviar_udp(puerto, mensaje, numero): """ Envía un paquete UDP y espera por una respuesta SACK por hasta 10 segundos, gestionando los errores de TIMEOUT y NO_ACK. """ fecha_envio = time.strftime("%d/%m/%Y %H:%M:%S") # siempre definido TIMEOUT_SACK = 10 # Tiempo de espera en segundos for intento in range(REINTENTOS_MAX): sock = None try: # 1. Crear el socket UDP sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # 2. Establecer un tiempo de espera de 10 segundos para la respuesta sock.settimeout(TIMEOUT_SACK) # 3. Enviar el mensaje t0 = time.time() sock.sendto(mensaje, (HOST, puerto)) # 4. Esperar la respuesta data, _ = sock.recvfrom(1024) data = data.decode(errors="ignore") # 5. Si la respuesta contiene "SACK", devolver éxito if "SACK" in data: # ACK recibido fecha_recepcion = time.strftime("%d/%m/%Y %H:%M:%S") delta = round(time.time() - t0, 2) return ("UDP", puerto, mensaje.decode(), numero, fecha_envio, fecha_recepcion, data.strip(), delta) except socket.timeout: # Capturar específicamente la excepción de tiempo de espera # No reintentamos, ya que el paquete se envió pero no hubo respuesta. logging.error(f"Error de tiempo de espera para SACK. No se recibió respuesta en {TIMEOUT_SACK}s.") return ("UDP", puerto, mensaje.decode(), numero, fecha_envio, "-", "TIMEOUT", -1) except Exception as e: # Capturar otros errores (como un puerto inaccesible) y reintentar logging.warning(f"Error al enviar/recibir paquete UDP (Intento {intento + 1}): {e}") continue finally: # Asegurarse de que el socket se cierre, sin importar el resultado if sock: sock.close() # 6. Si se agotaron los reintentos (por errores de conexión), devolver NO_ACK logging.error("Todos los reintentos de conexión fallaron.") return ("UDP", puerto, mensaje.decode(), numero, fecha_envio, "-", "NO_ACK", -1) # ====== MAIN ====== def main(): resultados = [] print(f"Iniciando test de puertos en {HOST}...") for inicio, fin in PUERTO_RANGOS: for puerto in range(inicio, fin+1): msg, numero = generar_mensaje() # TCP print(f"Probando TCP {puerto}...") res_tcp = enviar_tcp(puerto, msg, numero) resultados.append(res_tcp) # UDP #print(f"Probando UDP {puerto}...") #res_udp = enviar_udp(puerto, msg, numero) # resultados.append(res_udp) # ====== GUARDAR EN EXCEL ====== wb = Workbook() ws = wb.active ws.title = "Resultados" headers = ["protocolo","puerto","reporte","numero de mensaje", "fechaenvio","fecharecepcion","ack","delta"] ws.append(headers) for fila in resultados: ws.append(fila) # Ajustar ancho de columnas for i, col in enumerate(headers, 1): ws.column_dimensions[get_column_letter(i)].width = max(15, len(col) + 2) wb.save(OUTPUT_FILE) print(f"\n✅ Prueba finalizada. Resultados guardados en {os.path.abspath(OUTPUT_FILE)}") if __name__ == "__main__": main()