Análisis de Canales Encubiertos de Red en Malware
Resumen
El malware utiliza canales encubiertos para disfrazar la comunicación C2 y la exfiltración de datos dentro del tráfico de red que parece legítimo. El túnel DNS codifica datos en consultas y respuestas DNS (utilizado por herramientas como iodine, dnscat2 y familias de malware como FrameworkPOS). El túnel ICMP oculta datos en los payloads de solicitudes/respuestas echo (icmpsh, ptunnel). Los canales encubiertos HTTP incrustan datos C2 en cabeceras, cookies o imágenes esteganográficas. El abuso de protocolos explota protocolos permitidos para evadir firewalls. La detección de túneles DNS alcanza una tasa de recuerdo del 99%+ con enfoques modernos basados en ML, aunque la exfiltración de bajo rendimiento sigue siendo un desafío. Palo Alto Unit42 ha rastreado tres campañas importantes de túnel DNS (TrkCdn, SecShow, Savvy Seahorse) hasta 2024, lo que demuestra la continua prevalencia de la técnica.
Cuándo Usar
- Al investigar incidentes de seguridad que requieran analizar canales encubiertos de red en malware
- Al construir reglas de detección o consultas de caza de amenazas para este dominio
- Cuando los analistas del SOC necesitan procedimientos estructurados para este tipo de análisis
- Al validar la cobertura de monitoreo de seguridad para técnicas de ataque relacionadas
Requisitos Previos
- Python 3.9+ con
scapy,dpkt,dnslib - Wireshark/tshark para análisis de PCAP
- Zeek (anteriormente Bro) para monitoreo de red
- Infraestructura de registro de consultas DNS
- Comprensión de los protocolos DNS, ICMP y HTTP a nivel de paquete
Flujo de Trabajo
Paso 1: Detección de Túnel DNS
#!/usr/bin/env python3
"""Detecta túneles DNS y canales encubiertos en el tráfico de red."""
import sys
import json
import math
from collections import Counter, defaultdict
try:
from scapy.all import rdpcap, DNS, DNSQR, DNSRR, IP, ICMP
except ImportError:
print("pip install scapy")
sys.exit(1)
def entropy(data):
if not data:
return 0
freq = Counter(data)
length = len(data)
return -sum((c/length) * math.log2(c/length) for c in freq.values())
def analyze_dns_tunneling(pcap_path):
"""Detecta indicadores de túnel DNS en PCAP."""
packets = rdpcap(pcap_path)
domain_stats = defaultdict(lambda: {
"queries": 0, "total_qname_len": 0, "subdomain_lengths": [],
"query_types": Counter(), "unique_subdomains": set(),
})
for pkt in packets:
if pkt.haslayer(DNS) and pkt.haslayer(DNSQR):
qname = pkt[DNSQR].qname.decode('utf-8', errors='replace').rstrip('.')
qtype = pkt[DNSQR].qtype
parts = qname.split('.')
if len(parts) >= 3:
base_domain = '.'.join(parts[-2:])
subdomain = '.'.join(parts[:-2])
stats = domain_stats[base_domain]
stats["queries"] += 1
stats["total_qname_len"] += len(qname)
stats["subdomain_lengths"].append(len(subdomain))
stats["query_types"][qtype] += 1
stats["unique_subdomains"].add(subdomain)
# Puntúa dominios por indicadores de túnel
suspicious = []
for domain, stats in domain_stats.items():
if stats["queries"] < 5:
continue
avg_subdomain_len = (sum(stats["subdomain_lengths"]) /
len(stats["subdomain_lengths"]))
unique_ratio = len(stats["unique_subdomains"]) / stats["queries"]
# Calcula entropía de subdominios
all_subdomains = ''.join(stats["unique_subdomains"])
sub_entropy = entropy(all_subdomains)
score = 0
reasons = []
if avg_subdomain_len > 30:
score += 30
reasons.append(f"Subdominios largos (promedio {avg_subdomain_len:.0f} caracteres)")
if unique_ratio > 0.9:
score += 25
reasons.append(f"Alta unicidad ({unique_ratio:.2%})")
if sub_entropy > 4.0:
score += 25
reasons.append(f"Alta entropía ({sub_entropy:.2f})")
if stats["query_types"].get(16, 0) > 10: # Registros TXT
score += 20
reasons.append(f"Muchas consultas TXT ({stats['query_types'][16]})")
if score >= 50:
suspicious.append({
"domain": domain,
"score": score,
"queries": stats["queries"],
"avg_subdomain_length": round(avg_subdomain_len, 1),
"unique_subdomains": len(stats["unique_subdomains"]),
"subdomain_entropy": round(sub_entropy, 2),
"reasons": reasons,
})
return sorted(suspicious, key=lambda x: -x["score"])
def analyze_icmp_tunneling(pcap_path):
"""Detecta túnel ICMP en PCAP."""
packets = rdpcap(pcap_path)
icmp_stats = defaultdict(lambda: {"count": 0, "payload_sizes": [], "payloads": []})
for pkt in packets:
if pkt.haslayer(ICMP) and pkt.haslayer(IP):
src = pkt[IP].src
dst = pkt[IP].dst
key = f"{src}->{dst}"
payload = bytes(pkt[ICMP].payload)
icmp_stats[key]["count"] += 1
icmp_stats[key]["payload_sizes"].append(len(payload))
if len(payload) > 64:
icmp_stats[key]["payloads"].append(payload[:100])
suspicious = []
for flow, stats in icmp_stats.items():
if stats["count"] < 5:
continue
avg_size = sum(stats["payload_sizes"]) / len(stats["payload_sizes"])
if avg_size > 64 or stats["count"] > 100:
suspicious.append({
"flow": flow,
"packets": stats["count"],
"avg_payload_size": round(avg_size, 1),
"reason": "Payloads ICMP grandes/frecuentes sugieren túnel",
})
return suspicious
if __name__ == "__main__":
if len(sys.argv) < 2:
print(f"Uso: {sys.argv[0]} <archivo_pcap>")
sys.exit(1)
print("[+] Análisis de Túnel DNS")
dns_results = analyze_dns_tunneling(sys.argv[1])
for r in dns_results:
print(f" {r['domain']} (puntuación: {r['score']})")
for reason in r['reasons']:
print(f" - {reason}")
print("\n[+] Análisis de Túnel ICMP")
icmp_results = analyze_icmp_tunneling(sys.argv[1])
for r in icmp_results:
print(f" {r['flow']}: {r['reason']}")
Criterios de Validación
- Túnel DNS detectado mediante análisis de entropía, longitud de subdominio y volumen de consultas
- Canales encubiertos ICMP identificados por anomalías en el tamaño de payload
- Dominios de túnel distinguidos del tráfico legítimo de CDN/nube
- Volumen de exfiltración de datos estimado a partir del tráfico capturado
- Patrones de comunicación C2 e intervalos de beaconing extraídos