Construcción de Sistema de Seguimiento de Infraestructura Adversaria
Resumen
El seguimiento de infraestructura adversaria utiliza registros DNS pasivos, registros de transparencia de certificados, datos de registro WHOIS y enriquecimiento de IP para descubrir, mapear y monitorear redes de comando y control (C2) de actores de amenazas. Los atacantes frecuentemente reutilizan proveedores de alojamiento, registradores, certificados SSL y patrones de nombres en distintas campañas, lo que permite a los analistas pivotar desde indicadores conocidos para descubrir nueva infraestructura. Esta habilidad cubre la construcción de un sistema automatizado de seguimiento que identifica relaciones de infraestructura, detecta dominios recién registrados que coinciden con patrones de adversario y mantiene un mapa actualizado continuamente de las redes de actores de amenazas.
Cuándo Usar
- Al implementar o configurar capacidades de seguimiento de infraestructura adversaria en su entorno
- Al establecer controles de seguridad alineados con requisitos de cumplimiento
- Al construir o mejorar la arquitectura de seguridad para este dominio
- Al realizar evaluaciones de seguridad que requieran esta implementación
Prerrequisitos
- Python 3.9+ con las bibliotecas
requests,dnspython,python-whois,shodan,networkx - Claves de API: SecurityTrails, PassiveTotal/RiskIQ, Shodan, VirusTotal
- Acceso a fuentes de datos DNS pasivos
- Comprensión de infraestructura DNS, alojamiento y registro de dominios
- Base de datos de grafos (Neo4j) o NetworkX para visualización de relaciones
Conceptos Clave
DNS Pasivo
El DNS pasivo captura datos históricos de resolución DNS, registrando qué dominios se resolvieron a qué IPs y cuándo. A diferencia de las consultas DNS activas, el DNS pasivo conserva relaciones históricas incluso después de que los registros cambien, permitiendo a los analistas rastrear cambios en la infraestructura, identificar patrones de alojamiento compartido y descubrir dominios relacionados que se resolvieron a las mismas direcciones IP a lo largo del tiempo.
Pivote de Infraestructura
El pivote identifica infraestructura relacionada siguiendo conexiones: pivote por IP (encontrar todos los dominios en una IP), pivote por dominio (encontrar todas las IPs a las que un dominio se resolvió), pivote por WHOIS (encontrar dominios con el mismo registrante), pivote por certificado (encontrar hosts que comparten certificados SSL) y pivote por NS/MX (encontrar dominios que usan los mismos servidores de nombres o de correo).
Patrones de Infraestructura Adversaria
Los actores de amenazas exhiben patrones: registradores preferidos (Namecheap, REG.RU, Tucows), alojamiento preferido (proveedores de alojamiento a prueba de balas, servicios en la nube), algoritmos de generación de dominios (DGA), patrones de nombres consistentes y reutilización de certificados en distintas campañas.
Flujo de Trabajo
Paso 1: Descubrimiento de Infraestructura con DNS Pasivo
import requests
import json
from collections import defaultdict
from datetime import datetime
class InfrastructureTracker:
def __init__(self, securitytrails_key=None, vt_key=None, shodan_key=None):
self.st_key = securitytrails_key
self.vt_key = vt_key
self.shodan_key = shodan_key
self.infrastructure_graph = defaultdict(lambda: {"nodes": set(), "edges": []})
def passive_dns_lookup(self, domain):
"""Consultar DNS pasivo para el historial de resolución de un dominio."""
headers = {"apikey": self.st_key}
url = f"https://api.securitytrails.com/v1/history/{domain}/dns/a"
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
records = resp.json().get("records", [])
history = []
for record in records:
for value in record.get("values", []):
history.append({
"domain": domain,
"ip": value.get("ip", ""),
"first_seen": record.get("first_seen", ""),
"last_seen": record.get("last_seen", ""),
"type": record.get("type", "a"),
})
print(f"[+] DNS pasivo para {domain}: {len(history)} registros")
return history
return []
def reverse_ip_lookup(self, ip_address):
"""Encontrar todos los dominios alojados en una dirección IP."""
headers = {"apikey": self.st_key}
url = f"https://api.securitytrails.com/v1/ips/nearby/{ip_address}"
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
blocks = resp.json().get("blocks", [])
domains = []
for block in blocks:
for site in block.get("sites", []):
domains.append(site)
print(f"[+] IP inversa para {ip_address}: {len(domains)} dominios")
return domains
return []
def whois_lookup(self, domain):
"""Obtener datos de registro WHOIS para pivote."""
headers = {"apikey": self.st_key}
url = f"https://api.securitytrails.com/v1/domain/{domain}/whois"
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
data = resp.json()
whois_data = {
"domain": domain,
"registrar": data.get("registrar", ""),
"registrant_org": data.get("registrant_org", ""),
"registrant_email": data.get("registrant_email", ""),
"name_servers": data.get("nameServers", []),
"created_date": data.get("createdDate", ""),
"updated_date": data.get("updatedDate", ""),
"expires_date": data.get("expiresDate", ""),
}
return whois_data
return {}
def pivot_from_seed(self, seed_indicator, indicator_type="domain", depth=2):
"""Pivotar recursivamente desde un indicador semilla para descubrir infraestructura."""
discovered = {"domains": set(), "ips": set(), "relationships": []}
if indicator_type == "domain":
discovered["domains"].add(seed_indicator)
# Obtener IPs del dominio
pdns = self.passive_dns_lookup(seed_indicator)
for record in pdns:
ip = record["ip"]
discovered["ips"].add(ip)
discovered["relationships"].append({
"source": seed_indicator, "target": ip,
"type": "resolves_to",
"first_seen": record["first_seen"],
"last_seen": record["last_seen"],
})
if depth > 1:
# Búsqueda inversa en las IPs descubiertas
reverse_domains = self.reverse_ip_lookup(ip)
for rd in reverse_domains[:20]:
discovered["domains"].add(rd)
discovered["relationships"].append({
"source": rd, "target": ip,
"type": "hosted_on",
})
elif indicator_type == "ip":
discovered["ips"].add(seed_indicator)
domains = self.reverse_ip_lookup(seed_indicator)
for domain in domains[:20]:
discovered["domains"].add(domain)
discovered["relationships"].append({
"source": domain, "target": seed_indicator,
"type": "hosted_on",
})
print(f"[+] Pivote desde {seed_indicator}: "
f"{len(discovered['domains'])} dominios, "
f"{len(discovered['ips'])} IPs, "
f"{len(discovered['relationships'])} relaciones")
return discovered
tracker = InfrastructureTracker(
securitytrails_key="YOUR_ST_KEY",
vt_key="YOUR_VT_KEY",
)
Paso 2: Construir Grafo de Infraestructura
import networkx as nx
class InfrastructureGraph:
def __init__(self):
self.graph = nx.Graph()
def add_discovery(self, discovery_data):
"""Agregar infraestructura descubierta al grafo."""
for domain in discovery_data["domains"]:
self.graph.add_node(domain, type="domain")
for ip in discovery_data["ips"]:
self.graph.add_node(ip, type="ip")
for rel in discovery_data["relationships"]:
self.graph.add_edge(
rel["source"], rel["target"],
relationship=rel["type"],
first_seen=rel.get("first_seen", ""),
last_seen=rel.get("last_seen", ""),
)
def find_clusters(self):
"""Identificar clústeres de infraestructura."""
components = list(nx.connected_components(self.graph))
clusters = []
for component in components:
domains = [n for n in component if self.graph.nodes[n].get("type") == "domain"]
ips = [n for n in component if self.graph.nodes[n].get("type") == "ip"]
clusters.append({
"size": len(component),
"domains": sorted(domains),
"ips": sorted(ips),
"domain_count": len(domains),
"ip_count": len(ips),
})
clusters.sort(key=lambda x: x["size"], reverse=True)
print(f"[+] Clústeres de infraestructura: {len(clusters)}")
return clusters
def find_hub_nodes(self, top_n=10):
"""Encontrar nodos de alta centralidad (infraestructura compartida)."""
centrality = nx.degree_centrality(self.graph)
top_nodes = sorted(centrality.items(), key=lambda x: x[1], reverse=True)[:top_n]
hubs = []
for node, score in top_nodes:
hubs.append({
"node": node,
"type": self.graph.nodes[node].get("type", "unknown"),
"centrality": round(score, 4),
"connections": self.graph.degree(node),
})
return hubs
def export_graph(self, output_file="infrastructure_graph.json"):
data = nx.node_link_data(self.graph)
with open(output_file, "w") as f:
json.dump(data, f, indent=2)
print(f"[+] Grafo exportado: {self.graph.number_of_nodes()} nodos, "
f"{self.graph.number_of_edges()} aristas")
infra_graph = InfrastructureGraph()
discovery = tracker.pivot_from_seed("evil-domain.com", depth=2)
infra_graph.add_discovery(discovery)
clusters = infra_graph.find_clusters()
hubs = infra_graph.find_hub_nodes()
infra_graph.export_graph()
Paso 3: Monitorear Nueva Infraestructura
import time
class InfrastructureMonitor:
def __init__(self, tracker, known_indicators):
self.tracker = tracker
self.known = set(known_indicators)
self.alerts = []
def check_new_registrations(self, patterns):
"""Verificar dominios recién registrados que coincidan con patrones de adversario."""
import re
new_domains = []
for pattern in patterns:
# Consultar SecurityTrails por nuevos dominios que coincidan con el patrón
headers = {"apikey": self.tracker.st_key}
url = "https://api.securitytrails.com/v1/domains/list"
params = {"include_ips": "true", "page": 1}
body = {"filter": {"keyword": pattern}}
resp = requests.post(url, headers=headers, json=body, timeout=30)
if resp.status_code == 200:
records = resp.json().get("records", [])
for record in records:
domain = record.get("hostname", "")
if domain not in self.known:
new_domains.append({
"domain": domain,
"pattern_matched": pattern,
"first_seen": datetime.now().isoformat(),
})
self.known.add(domain)
if new_domains:
print(f"[ALERTA] {len(new_domains)} nuevos dominios coinciden con patrones")
self.alerts.extend(new_domains)
return new_domains
def generate_infrastructure_report(self, clusters, hubs):
report = f"""# Informe de Seguimiento de Infraestructura Adversaria
Generado: {datetime.now().isoformat()}
## Resumen
- Clústeres de infraestructura identificados: {len(clusters)}
- Total de dominios rastreados: {sum(c['domain_count'] for c in clusters)}
- Total de IPs rastreadas: {sum(c['ip_count'] for c in clusters)}
- Nuevos dominios detectados: {len(self.alerts)}
## Principales Hubs de Infraestructura
| Nodo | Tipo | Conexiones | Centralidad |
|------|------|------------|-------------|
"""
for hub in hubs[:10]:
report += (f"| {hub['node']} | {hub['type']} "
f"| {hub['connections']} | {hub['centrality']} |\n")
report += "\n## Clústeres de Infraestructura\n"
for i, cluster in enumerate(clusters[:5], 1):
report += f"\n### Cluster {i} ({cluster['size']} nodos)\n"
report += f"- Dominios: {', '.join(cluster['domains'][:5])}\n"
report += f"- IPs: {', '.join(cluster['ips'][:5])}\n"
with open("infrastructure_report.md", "w") as f:
f.write(report)
print("[+] Informe de infraestructura guardado")
monitor = InfrastructureMonitor(tracker, known_indicators=set())
Criterios de Validación
- Las consultas de DNS pasivo devuelven datos históricos de resolución
- Las búsquedas inversas de IP descubren dominios co-alojados
- El pivote de infraestructura se expande desde indicadores semilla
- El análisis de grafos identifica clústeres y nodos hub
- Se detecta nueva infraestructura mediante monitoreo de patrones
- Se generan informes con recomendaciones accionables