Problema

Los profesionales que realizan pruebas de penetración o auditorías de Active Directory (AD) suelen pasar de una herramienta a otra: primero descubren controladores de dominio vía DNS, después consultan LDAP, luego prueban Kerberos, ejecutan scripts de enumeración y finalmente generan informes. Cada fase requiere lanzar un binario, interpretar su salida y pasar datos manualmente al siguiente paso. El flujo fragmentado genera:

  • Cambios de contexto que aumentan la probabilidad de errores.
  • Dificultad para reproducir una misma metodología en diferentes entornos.
  • Falta de control centralizado sobre el alcance autorizado, lo que puede provocar ejecuciones accidentales fuera de los límites del proyecto.

En entornos donde el tiempo de respuesta es crítico (incidentes, pruebas de red team) o donde se necesita una metodología estandarizada (cumplimiento, auditorías internas), este patrón de trabajo fragmentado se vuelve un cuello de botella.

Causa

  1. Herramientas aisladas – La mayoría de los utilitarios de AD (BloodHound, impacket, nmap, PowerView, etc.) están diseñados como ejecutables independientes que exportan texto o CSV. No comparten un formato de salida común.
  2. Falta de capa de orquestación – No existe un motor que reciba la salida de una herramienta, la transforme y la alimente a la siguiente sin intervención humana.
  3. Gestión de alcance manual – Los scripts tradicionales dependen de variables de entorno o argumentos que el operador debe validar a mano antes de ejecutar cualquier acción.
  4. Persistencia de credenciales – Muchos pipelines guardan contraseñas en texto plano o en logs, lo que rompe la política de “no almacenar credenciales en claro”.
  5. Timeout y manejo de errores dispersos – Cada herramienta gestiona sus propios timeouts; si una falla, el resto del pipeline queda colgado o produce resultados incompletos.

Solución

Crear una capa de orquestación basada en Python que:

  • Descubra y normalice la infraestructura AD mediante módulos reutilizables (DNS SRV, LDAP RootDSE, SMB).
  • Ejecute herramientas externas como subprocesses, capture su salida en JSON y la convierta a un esquema interno.
  • Valide el alcance antes de cualquier acción sensible: dry‑run → confirmación explícita → ejecución.
  • Aplique políticas de seguridad: elimina credenciales del entorno después de usarlas, evita logs de texto plano y fuerza TLS/SSL cuando el objetivo lo soporta.
  • Permita paralelismo mediante concurrent.futures para acelerar la fase de enumeración sin perder control.
  • Exporte resultados en un formato JSON estructurado que pueda ser consumido por SIEM, dashboards o herramientas de correlación.

Arquitectura mínima

└─ orchestrator.py
   ├─ discovery/
   │   ├─ dns_srv.py
   │   ├─ ldap_rootdse.py
   │   └─ smb_enum.py
   ├─ assessment/
   │   ├─ kerberos_check.py
   │   ├─ bloodhound_collect.py
   │   └─ custom_tool_wrapper.py
   ├─ utils/
   │   ├─ json_schema.py
   │   ├─ scope_validator.py
   │   └─ credential_manager.py
   └─ main.py

Cada módulo expone una función run(target, creds, **options) que devuelve un diccionario compatible con el esquema global. El main.py orquesta el flujo siguiendo los pasos de validación, ejecución paralela y consolidación de resultados.

Implementación práctica

  1. Validación de alcance – Un pequeño script que lee un archivo scope.yaml y compara el objetivo contra la lista autorizada. Si no coincide, aborta.
  2. Ejecución con timeout – Usa subprocess.run(..., timeout=60) para evitar bloqueos.
  3. Normalización – Un wrapper que traduce la salida CSV de BloodHound a JSON bajo la clave bloodhound.
  4. Persistencia segura – El módulo credential_manager carga credenciales desde un vault (por ejemplo, keyring) y las elimina de la memoria tras su uso.

Cuándo aplicar esta solución

  • Auditorías repetitivas – Cuando la organización necesita ejecutar la misma serie de pruebas en varios dominios o en entornos de pruebas y producción.
  • Equipos de red/blue team – Cuando la velocidad de ejecución y la trazabilidad del alcance son críticas.
  • Entornos con políticas de cero credenciales en logs – Cuando la normativa prohíbe almacenar contraseñas en texto plano.
  • Integración CI/CD de seguridad – Cuando se desea lanzar escaneos de AD como parte de un pipeline de despliegue.

No es adecuada si:

  • Sólo se necesita una única herramienta puntual y la sobrecarga de la orquestación supera el beneficio.
  • El entorno carece de Python o de permisos para ejecutar subprocesses con privilegios elevados.

Código

#!/usr/bin/env python3
import json, subprocess, concurrent.futures, pathlib, sys
from utils.scope_validator import validate_scope
from utils.credential_manager import get_credentials, clear_credentials

TARGET = sys.argv[1]          # ej. "corp.example.com"
SCOPE_FILE = "scope.yaml"

if not validate_scope(TARGET, SCOPE_FILE):
    sys.exit(f"[!] {TARGET} no está dentro del alcance autorizado")

creds = get_credentials()     # devuelve dict con user y password

def run_tool(command):
    try:
        result = subprocess.run(
            command,
            capture_output=True,
            text=True,
            timeout=90,
            env={"KRB5CCNAME": "/tmp/krb5cc_temp"}
        )
        return json.loads(result.stdout) if result.stdout else {}
    except subprocess.TimeoutExpired:
        return {"error": "timeout"}
    except json.JSONDecodeError:
        return {"error": "invalid json"}

tools = [
    ["python3", "discovery/dns_srv.py", TARGET],
    ["python3", "discovery/ldap_rootdse.py", TARGET],
    ["python3", "assessment/kerberos_check.py", TARGET, creds["user"], creds["password"]],
    ["bloodhound", "-d", TARGET, "-c", "All", "-j", "output.json"]
]

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(run_tool, cmd): cmd for cmd in tools}
    aggregated = {}
    for future in concurrent.futures.as_completed(futures):
        cmd = futures[future]
        aggregated[" ".join(cmd)] = future.result()

clear_credentials(creds)

output_path = pathlib.Path("assessment_results.json")
output_path.write_text(json.dumps(aggregated, indent=2))
print(f"[+] Resultados guardados en {output_path}")

Verificación

  1. Ejecutar el script con un objetivo que esté en scope.yaml. El proceso debe terminar en menos de 5 min y generar assessment_results.json.
  2. Abrir el JSON y comprobar que cada clave corresponde a una herramienta y que los campos error están ausentes.
  3. Verificar que el archivo de credenciales temporal (/tmp/krb5cc_temp) haya sido eliminado tras la ejecución (ls /tmp/krb5cc_temp debe fallar).
  4. Simular un objetivo fuera de alcance; el script debe abortar antes de lanzar cualquier subprocess.

Notas adicionales

  • Manejo de TLS – Cuando la herramienta lo permite, agrega --starttls o -s para forzar la capa segura. La mayoría de los módulos LDAP en Python aceptan use_ssl=True.
  • Escalado – En entornos con cientos de dominios, sustituye ThreadPoolExecutor por ProcessPoolExecutor para evitar la limitación del GIL en operaciones intensivas de I/O.
  • Extensibilidad – Cada nuevo binario solo necesita un wrapper que devuelva JSON. Mantener un esquema versionado evita rupturas al añadir campos.
  • Mapeo ATT&CK – Añadir una capa que asocie cada hallazgo con una técnica MITRE facilita la generación de reportes de riesgo y la priorización de mitigaciones.
  • Auditoría – Registra en un archivo separado los hashes de los binarios ejecutados y sus versiones; esto ayuda a reproducir exactamente la misma cadena de herramientas en auditorías posteriores.