Script Python para el Análisis de DNS de Dominios y la Conectividad Web

La gestión de infraestructuras web a menudo requiere verificar la resolución DNS de dominios públicos y su disponibilidad. Un script como el que se presenta a continuación puede ser una herrramienta invaluable para administradores de sistemas y desarrolladores, permitiendo auditar un listado de URLs para obtener sus direcciones IP resueltas y el código de estado HTTP de sus servidores.

Configuración Inicial y Dependencias

Para ejecutar los scripts, es necesario instalar varias bibliotecas de Python. Utiliza pip en tu terminal para instalarlas:

pip install requests dnspython aiohttp
  • requests: Para realizar peticiones HTTP de forma síncrona y en hilos.
  • dnspython: Para consultas DNS.
  • aiohttp: Para peticiones HTTP asíncronas.

Preparación del Archivo de Entrada: urls_a_verificar.txt

El script requiere un archivo de texto llamado urls_a_verificar.txt en el mismo directorio donde se ejecuta. Este archivo debe contener una URL por línea, la cual será analizada. Las líneas que comiencen con # serán ignoradas.

Si el archivo no existe, el script lo creará automáticamente con un conjunto de ejemplos. A continuación, se muestra un ejemplo del contenido del archivo:

# URLs de ejemplo para verificación
https://www.google.com
https://www.example.org:443
http://www.nonexistentdomain.xyz:80
https://api.github.com

Implementación Síncrona del Verificador

Esta primera versión del script procesa cada URL de forma secuencial. Realiza una resolución DNS para obtener las direcciones IP del dominio y luego intenta una petición HTTP para verificar la disponibilidad del sitio, capturando el código de estado. Los resultados se guardan en un archivo CSV.

import os
import csv
import dns.resolver
import requests
from urllib.parse import urlparse

# Cabeceras HTTP para simular una petición de navegador
cabeceras_http = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
    'Accept-Language': 'es-ES,es;q=0.9',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1',
    'Sec-Fetch-Dest': 'document',
    'Sec-Fetch-Mode': 'navigate',
    'Sec-Fetch-Site': 'none',
    'Sec-Fetch-User': '?1'
}

def inicializar_archivo_urls(nombre_archivo="urls_a_verificar.txt"):
    """Crea el archivo de URLs de ejemplo si no existe."""
    ruta_completa = os.path.join(os.getcwd(), nombre_archivo)
    contenido_ejemplo = """# Ingresa una URL por línea. Las líneas que inician con '#' son ignoradas.
https://www.google.com
https://www.python.org:443
http://www.ejemplo.org:80
https://mail.google.com
"""
    if not os.path.exists(ruta_completa):
        with open(ruta_completa, 'w', encoding='utf-8') as f:
            f.write(contenido_ejemplo)
        print(f"Archivo '{nombre_archivo}' creado automáticamente en: {ruta_completa}")
        print("Por favor, revisa y edita el archivo con tus URLs antes de continuar.")
        input("Presiona Enter para salir y editar el archivo.")
        exit()
    print(f"Utilizando el archivo '{nombre_archivo}' ubicado en: {ruta_completa}")
    return ruta_completa

def leer_urls_del_archivo(ruta_archivo):
    """Lee las URLs desde el archivo, ignorando comentarios."""
    with open(ruta_archivo, 'r', encoding='utf-8') as f:
        return [line.strip() for line in f if line.strip() and not line.startswith('#')]

def ejecutar_verificacion_sincrona(lista_urls, ruta_salida_csv="resultados_verificacion_sincrona.csv"):
    """Realiza la verificación DNS y HTTP de forma síncrona."""
    with open(ruta_salida_csv, 'w', newline='', encoding='utf-8') as archivo_csv:
        escritor_csv = csv.writer(archivo_csv)
        escritor_csv.writerow(['URL Solicitada', 'Dominio Analizado', 'IPs Resueltas (DNS)', 'Estado HTTP / Error'])

        for url_actual in lista_urls:
            print(f"Procesando: {url_actual}")
            nombre_dominio = ""
            direcciones_ip = "N/A"
            estado_http = "N/A"

            try:
                url_analizada = urlparse(url_actual)
                # Extraer el dominio sin el puerto para la consulta DNS
                nombre_dominio = url_analizada.netloc.split(':')[0] if url_analizada.netloc else ""

                if not nombre_dominio:
                    estado_http = "URL inválida o sin dominio"
                    escritor_csv.writerow([url_actual, nombre_dominio, direcciones_ip, estado_http])
                    continue

                # Resolución DNS para registros A
                respuestas_dns = dns.resolver.resolve(nombre_dominio, 'A')
                direcciones_ip = ', '.join(str(r.address) for r in respuestas_dns)

                # Petición HTTP
                try:
                    respuesta_http = requests.get(url_actual, headers=cabeceras_http, timeout=10, allow_redirects=True)
                    estado_http = respuesta_http.status_code
                except requests.exceptions.RequestException as e:
                    estado_http = f"Error HTTP: {type(e).__name__} - {str(e)}"

            except dns.resolver.NoAnswer:
                direcciones_ip = "No se encontraron registros A"
            except dns.resolver.NXDOMAIN:
                direcciones_ip = "Dominio no existente (NXDOMAIN)"
            except Exception as e:
                estado_http = f"Error General: {type(e).__name__} - {str(e)}"
            finally:
                escritor_csv.writerow([url_actual, nombre_dominio, direcciones_ip, estado_http])
    print(f"\nVerificación síncrona completada. Resultados en '{ruta_salida_csv}'")

# --- Lógica principal para la ejecución síncrona ---
if __name__ == "__main__":
    ruta_archivo_urls = inicializar_archivo_urls()
    lista_urls_a_procesar = leer_urls_del_archivo(ruta_archivo_urls)
    if not lista_urls_a_procesar:
        print("El archivo de URLs está vacío o solo contiene comentarios. Saliendo.")
        exit()
    input("\nPresiona Enter para iniciar la verificación síncrona...")
    ejecutar_verificacion_sincrona(lista_urls_a_procesar)
    input("Presiona Enter para finalizar el programa.")

Manejo de Errores Comunes

Durante la ejecución, pueden surgir diversos errores que son capturados y registrados en el archivo CSV. Algunos de los más frecuentes incluyen:

  • Errores de DNS:
    • dns.resolver.NoAnswer: El servidor DNS respondió que no hay registros del tipo 'A' para el dominio.
    • dns.resolver.NXDOMAIN: El dominio no existe o no pudo ser resuelto por el servidor DNS.
  • Errores de Conexión HTTP:
    • requests.exceptions.ConnectTimeout: La conexión al servidor web excedió el tiempo límite especificado. Esto puede indicar un servidor inactivo, problemas de red o un cortafuegos.
    • requests.exceptions.ConnectionError: No se pudo establecer una conexión con el servidor. Esto podría deberse a un servidor que activamente rechazó la conexión (por ejemplo, si no hay un servicio HTTP/S escuchando en ese puerto o un cortafuegos lo bloquea).
    • requests.exceptions.SSLError: Problemas con el certificado SSL/TLS del sitio, como un certificado caducado, un nombre de host no coincidente o una cadena de certificados incompleta. Aunque el sitio pueda responder, la conexión segura no es válida.
    • requests.exceptions.ReadTimeout: El servidor web aceptó la conexión, pero no envió datos en el tiempo esperado.

Mejoras de Rendimiento: Concurrencia

Para acelerar el proceso cuando se manejan muchas URLs, se pueden utilizar enfoques concurrentes: multihilo o asíncrono.

Implementación Multihilo

La versión multihilo utiliza concurrent.futures.ThreadPoolExecutor para ejecutar múltiples verificaciones simultáneamente, aprovechando los núcleos del procesador de manera más eficiente para tareas I/O-bound (como peticiones de red).

import os
import csv
import dns.resolver
import requests
from urllib.parse import urlparse
import concurrent.futures

# Cabeceras HTTP (asegúrate de que estas estén definidas o se importen si están en otro archivo)
cabeceras_http = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
    'Accept-Language': 'es-ES,es;q=0.9',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1',
    'Sec-Fetch-Dest': 'document',
    'Sec-Fetch-Mode': 'navigate',
    'Sec-Fetch-Site': 'none',
    'Sec-Fetch-User': '?1'
}

def inicializar_archivo_urls(nombre_archivo="urls_a_verificar.txt"):
    """Crea el archivo de URLs de ejemplo si no existe."""
    ruta_completa = os.path.join(os.getcwd(), nombre_archivo)
    contenido_ejemplo = """# Ingresa una URL por línea. Las líneas que inician con '#' son ignoradas.
https://www.google.com
https://www.python.org:443
http://www.ejemplo.org:80
https://mail.google.com
"""
    if not os.path.exists(ruta_completa):
        with open(ruta_completa, 'w', encoding='utf-8') as f:
            f.write(contenido_ejemplo)
        print(f"Archivo '{nombre_archivo}' creado automáticamente en: {ruta_completa}")
        print("Por favor, revisa y edita el archivo con tus URLs antes de continuar.")
        input("Presiona Enter para salir y editar el archivo.")
        exit()
    print(f"Utilizando el archivo '{nombre_archivo}' ubicado en: {ruta_completa}")
    return ruta_completa

def leer_urls_del_archivo(ruta_archivo):
    """Lee las URLs desde el archivo, ignorando comentarios."""
    with open(ruta_archivo, 'r', encoding='utf-8') as f:
        return [line.strip() for line in f if line.strip() and not line.startswith('#')]

def procesar_url_hilo(url_a_revisar):
    """Función para ser ejecutada por cada hilo: resuelve DNS y verifica HTTP."""
    nombre_dominio = ""
    direcciones_ip = "N/A"
    estado_http = "N/A"

    try:
        url_analizada = urlparse(url_a_revisar)
        nombre_dominio = url_analizada.netloc.split(':')[0] if url_analizada.netloc else ""

        if not nombre_dominio:
            estado_http = "URL inválida o sin dominio"
            return url_a_revisar, nombre_dominio, direcciones_ip, estado_http

        respuestas_dns = dns.resolver.resolve(nombre_dominio, 'A')
        direcciones_ip = ', '.join(str(r.address) for r in respuestas_dns)

        try:
            respuesta_http = requests.get(url_a_revisar, headers=cabeceras_http, timeout=10, allow_redirects=True)
            estado_http = respuesta_http.status_code
        except requests.exceptions.RequestException as e:
            estado_http = f"Error HTTP: {type(e).__name__} - {str(e)}"

    except dns.resolver.NoAnswer:
        direcciones_ip = "No se encontraron registros A"
    except dns.resolver.NXDOMAIN:
        direcciones_ip = "Dominio no existente (NXDOMAIN)"
    except Exception as e:
        estado_http = f"Error General: {type(e).__name__} - {str(e)}"
    
    return url_a_revisar, nombre_dominio, direcciones_ip, estado_http

def ejecutar_verificacion_multihilo(lista_urls, max_hilos=10, ruta_salida_csv="resultados_verificacion_multihilo.csv"):
    """Realiza la verificación DNS y HTTP de forma concurrente usando hilos."""
    with open(ruta_salida_csv, 'w', newline='', encoding='utf-8') as archivo_csv:
        escritor_csv = csv.writer(archivo_csv)
        escritor_csv.writerow(['URL Solicitada', 'Dominio Analizado', 'IPs Resueltas (DNS)', 'Estado HTTP / Error'])

        with concurrent.futures.ThreadPoolExecutor(max_workers=max_hilos) as ejecutor_hilos:
            # Mapear la función de procesamiento a cada URL
            futuros_por_url = {ejecutor_hilos.submit(procesar_url_hilo, url_item): url_item for url_item in lista_urls}

            for futuro_tarea in concurrent.futures.as_completed(futuros_por_url):
                url_procesada = futuros_por_url[futuro_tarea]
                try:
                    resultado = futuro_tarea.result()
                    escritor_csv.writerow(resultado)
                    print(f"Completado: {url_procesada}")
                except Exception as exc:
                    # Este bloque captura excepciones de la función procesar_url_hilo
                    # Sin embargo, procesar_url_hilo ya captura la mayoría y devuelve un estado de error.
                    # Esto sería para errores inesperados en el executor mismo.
                    print(f"Error inesperado al procesar {url_procesada}: {exc}")
                    escritor_csv.writerow([url_procesada, "N/A", "N/A", f"Error Executor: {str(exc)}"])
    print(f"\nVerificación multihilo completada. Resultados en '{ruta_salida_csv}'")

# --- Lógica principal para la ejecución multihilo ---
if __name__ == "__main__":
    ruta_archivo_urls = inicializar_archivo_urls()
    lista_urls_a_procesar = leer_urls_del_archivo(ruta_archivo_urls)
    if not lista_urls_a_procesar:
        print("El archivo de URLs está vacío o solo contiene comentarios. Saliendo.")
        exit()
    input("\nPresiona Enter para iniciar la verificación multihilo...")
    ejecutar_verificacion_multihilo(lista_urls_a_procesar)
    input("Presiona Enter para finalizar el programa.")

Implementación Asíncrona con aiohttp

La programación asíncrona, utilizando asyncio y aiohttp, es ideal para tareas de I/O-bound intensivas, como realizar múltiples peticiones de red. Permite que el programa realice otras operaciones mientras espera respuestas de red, lo que puede resultar en un rendimiento superior en comparación con la multihilo para este tipo de cargas de trabajo.

import os
import csv
import dns.resolver
import asyncio
import aiohttp
from urllib.parse import urlparse

# Cabeceras HTTP (asegúrate de que estas estén definidas o se importen si están en otro archivo)
cabeceras_http = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
    'Accept-Language': 'es-ES,es;q=0.9',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1',
    'Sec-Fetch-Dest': 'document',
    'Sec-Fetch-Mode': 'navigate',
    'Sec-Fetch-Site': 'none',
    'Sec-Fetch-User': '?1'
}

def inicializar_archivo_urls(nombre_archivo="urls_a_verificar.txt"):
    """Crea el archivo de URLs de ejemplo si no existe."""
    ruta_completa = os.path.join(os.getcwd(), nombre_archivo)
    contenido_ejemplo = """# Ingresa una URL por línea. Las líneas que inician con '#' son ignoradas.
https://www.google.com
https://www.python.org:443
http://www.ejemplo.org:80
https://mail.google.com
"""
    if not os.path.exists(ruta_completa):
        with open(ruta_completa, 'w', encoding='utf-8') as f:
            f.write(contenido_ejemplo)
        print(f"Archivo '{nombre_archivo}' creado automáticamente en: {ruta_completa}")
        print("Por favor, revisa y edita el archivo con tus URLs antes de continuar.")
        input("Presiona Enter para salir y editar el archivo.")
        exit()
    print(f"Utilizando el archivo '{nombre_archivo}' ubicado en: {ruta_completa}")
    return ruta_completa

def leer_urls_del_archivo(ruta_archivo):
    """Lee las URLs desde el archivo, ignorando comentarios."""
    with open(ruta_archivo, 'r', encoding='utf-8') as f:
        return [line.strip() for line in f if line.strip() and not line.startswith('#')]


async def obtener_estado_http(sesion_http, url_objetivo):
    """Realiza una petición HTTP asíncrona."""
    try:
        async with sesion_http.get(url_objetivo, headers=cabeceras_http, timeout=10, allow_redirects=True) as respuesta_http:
            return respuesta_http.status
    except aiohttp.ClientError as e:
        return f"Error HTTP Async: {type(e).__name__} - {str(e)}"
    except asyncio.TimeoutError:
        return "Error HTTP Async: Timeout de conexión"
    except Exception as e:
        return f"Error HTTP Async General: {type(e).__name__} - {str(e)}"

async def procesar_url_async(sesion_http, url_a_revisar):
    """Función asíncrona para resolver DNS y verificar HTTP para una URL."""
    nombre_dominio = ""
    direcciones_ip = "N/A"
    estado_http = "N/A"

    try:
        url_analizada = urlparse(url_a_revisar)
        nombre_dominio = url_analizada.netloc.split(':')[0] if url_analizada.netloc else ""

        if not nombre_dominio:
            estado_http = "URL inválida o sin dominio"
            return url_a_revisar, nombre_dominio, direcciones_ip, estado_http

        # dns.resolver no es nativamente asíncrono, se ejecuta de forma bloqueante o en un ThreadPool
        # Para simplificar, se mantiene síncrono aquí, pero podría envolverse en loop.run_in_executor
        respuestas_dns = dns.resolver.resolve(nombre_dominio, 'A')
        direcciones_ip = ', '.join(str(r.address) for r in respuestas_dns)

        estado_http = await obtener_estado_http(sesion_http, url_a_revisar)

    except dns.resolver.NoAnswer:
        direcciones_ip = "No se encontraron registros A"
    except dns.resolver.NXDOMAIN:
        direcciones_ip = "Dominio no existente (NXDOMAIN)"
    except Exception as e:
        estado_http = f"Error General Async: {type(e).__name__} - {str(e)}"
    
    return url_a_revisar, nombre_dominio, direcciones_ip, estado_http

async def ejecutar_verificacion_async(lista_urls, ruta_salida_csv="resultados_verificacion_async.csv"):
    """Orquesta la verificación asíncrona de un listado de URLs."""
    with open(ruta_salida_csv, 'w', newline='', encoding='utf-8') as archivo_csv:
        escritor_csv = csv.writer(archivo_csv)
        escritor_csv.writerow(['URL Solicitada', 'Dominio Analizado', 'IPs Resueltas (DNS)', 'Estado HTTP / Error'])

        async with aiohttp.ClientSession() as sesion_http:
            tareas_pendientes = [procesar_url_async(sesion_http, url_item) for url_item in lista_urls]
            for i, futura_tarea in enumerate(asyncio.as_completed(tareas_pendientes)):
                print(f"Progreso: {i+1}/{len(lista_urls)} URLs procesadas.")
                resultado = await futura_tarea
                escritor_csv.writerow(resultado)
    print(f"\nVerificación asíncrona completada. Resultados en '{ruta_salida_csv}'")

# --- Lógica principal para la ejecución asíncrona ---
if __name__ == "__main__":
    ruta_archivo_urls = inicializar_archivo_urls()
    lista_urls_a_procesar = leer_urls_del_archivo(ruta_archivo_urls)
    if not lista_urls_a_procesar:
        print("El archivo de URLs está vacío o solo contiene comentarios. Saliendo.")
        exit()
    input("\nPresiona Enter para iniciar la verificación asíncrona...")
    asyncio.run(ejecutar_verificacion_async(lista_urls_a_procesar))
    input("Presiona Enter para finalizar el programa.")

Formato de Salida CSV

Todos los métodos de verificación generan un archivo CSV con el siguiente formato de columnas:

  • URL Solicitada: La URL completa que se intentó verificar.
  • Dominio Analizado: El nombre de dominio extraído de la URL, utilizado para la consulta DNS.
  • IPs Resueltas (DNS): Una lista de direcciones IP obtenidas de la resolución DNS para el dominio, separadas por comas. Si no hay resolución, se indicará el error.
  • Estado HTTP / Error: El código de estado HTTP (por ejemplo, 200 para OK, 404 para No Encontrado) o una descripción detallada del error si la petición falló (timeout, error SSL, etc.).

Etiquetas: dns HTTP PythonRequests dnspython aiohttp

Publicado el 7-1 18:07

Etiquetas populares

©2026 Friki Work