3.4 Python para hacking
Python es el lenguaje de hacking por defecto no porque sea el más rápido o el más elegante, sino porque tiene una librería para todo, funciona en cualquier sistema y el tiempo entre "tengo una idea" y "está funcionando" es mínimo. Esta sección no es un tutorial de Python — asume que ya sabes escribir funciones y bucles. El objetivo es enseñar los patrones específicos que aparecen una y otra vez en scripts ofensivos: cómo envolver herramientas externas, cómo paralelizar sin que el código se convierta en un desastre, cómo construir paquetes de red desde cero y cómo automatizar interacciones web.
3.4.1 argparse + subprocess: wrapper profesional de herramientas
Un wrapper es un script Python que llama a otra herramienta (nmap, gobuster, ffuf...) con los argumentos correctos, procesa su salida y hace algo útil con ella. Es la forma más rápida de integrar herramientas existentes en flujos de trabajo automatizados sin reescribirlas desde cero.
argparse: argumentos de línea de comandos
import argparse
parser = argparse.ArgumentParser(
description="Scanner de puertos con nmap",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="Ejemplo: %(prog)s -t 192.168.1.1 -p 22,80,443 -v"
)
# Argumento posicional (obligatorio, sin --)
parser.add_argument("objetivo",
help="IP o rango CIDR del objetivo")
# Argumento opcional con valor
parser.add_argument("-p", "--puertos",
default="1-1024",
help="Puertos a escanear (defecto: 1-1024)")
# Argumento con tipo y validación
parser.add_argument("-w", "--workers",
type=int,
default=10,
choices=range(1, 51), # solo 1-50
metavar="N",
help="Hilos paralelos (1-50, defecto: 10)")
# Flag booleano
parser.add_argument("-v", "--verbose",
action="store_true",
help="Salida detallada")
# Argumento que puede repetirse: -x a -x b → ["a", "b"]
parser.add_argument("-x", "--excluir",
action="append",
metavar="PUERTO",
help="Puerto a excluir (puede usarse varias veces)")
# Argumento de archivo
parser.add_argument("-o", "--output",
type=argparse.FileType("w"),
default="-", # "-" = stdout
help="Archivo de salida (defecto: stdout)")
args = parser.parse_args()
print(f"Objetivo: {args.objetivo}")
print(f"Puertos: {args.puertos}")
print(f"Verbose: {args.verbose}")
subprocess: ejecutar y controlar procesos externos
import subprocess
# subprocess.run: la forma moderna y segura
# Pasar argumentos como lista (no como string) evita shell injection
resultado = subprocess.run(
["nmap", "-sV", "--open", args.objetivo, "-p", args.puertos],
capture_output=True, # capturar stdout y stderr
text=True, # decodificar a str en vez de bytes
timeout=120 # matar el proceso si tarda más de 2 min
)
if resultado.returncode != 0:
print(f"Error: {resultado.stderr}", file=sys.stderr)
sys.exit(1)
print(resultado.stdout)
# Filtrar solo las líneas de puertos abiertos
for linea in resultado.stdout.splitlines():
if "open" in linea:
print(linea)
# Popen: para procesos de larga duración con salida en tiempo real
proceso = subprocess.Popen(
["nmap", "-sV", "--open", "-p-", args.objetivo],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# Leer la salida línea a línea mientras el proceso corre
for linea in proceso.stdout:
linea = linea.rstrip()
if linea:
print(linea)
if "open" in linea:
args.output.write(linea + "\n")
proceso.wait() # esperar a que termine
Wrapper completo: nmap → lista de objetivos para otras herramientas
#!/usr/bin/env python3
"""
wrapper_nmap.py — escanea una red y genera listas de objetivos
por servicio: ssh.txt, http.txt, smb.txt...
"""
import argparse
import subprocess
import sys
import re
from pathlib import Path
from collections import defaultdict
def parsear_args():
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("red", help="IP o rango CIDR (ej: 192.168.1.0/24)")
p.add_argument("-o", "--outdir", default="./recon", help="Directorio de salida")
p.add_argument("-p", "--puertos", default="22,80,443,445,3389,8080")
return p.parse_args()
def escanear(red, puertos):
cmd = ["nmap", "-sV", "--open", "-p", puertos, red, "-oG", "-"]
resultado = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
if resultado.returncode != 0:
sys.exit(f"nmap falló: {resultado.stderr}")
return resultado.stdout
def parsear_grepeable(salida):
"""Devuelve {ip: [(puerto, servicio), ...]}"""
hosts = defaultdict(list)
for linea in salida.splitlines():
if not linea.startswith("Host:"):
continue
ip = linea.split()[1]
for m in re.finditer(r"(\d+)/open/tcp//(\w+)", linea):
puerto, servicio = m.group(1), m.group(2)
hosts[ip].append((int(puerto), servicio))
return hosts
def main():
args = parsear_args()
outdir = Path(args.outdir)
outdir.mkdir(parents=True, exist_ok=True)
print(f"[*] Escaneando {args.red}...")
salida = escanear(args.red, args.puertos)
hosts = parsear_grepeable(salida)
listas = defaultdict(list)
for ip, servicios in hosts.items():
for puerto, svc in servicios:
listas[svc].append(f"{ip}:{puerto}")
if args.verbose if hasattr(args, "verbose") else False:
print(f" {ip}:{puerto} → {svc}")
for svc, objetivos in listas.items():
ruta = outdir / f"{svc}.txt"
ruta.write_text("\n".join(objetivos) + "\n")
print(f"[+] {svc}: {len(objetivos)} objetivos → {ruta}")
if __name__ == "__main__":
main()
3.4.2 asyncio y threading para escaneos paralelos caseros
Escanear puertos o fuzzear directorios de forma secuencial tarda demasiado. La paralelización es imprescindible, pero hacerla mal (demasiados hilos, sin límites, sin control de errores) crashea el script o tumba el objetivo. Hay dos modelos: threading para operaciones que bloquean y asyncio para operaciones de I/O intensivo.
threading y ThreadPoolExecutor
import socket
import concurrent.futures
from threading import Lock
lock = Lock() # para escritura thread-safe en variables compartidas
abiertos = []
def probar_puerto(ip, puerto, timeout=1):
"""Devuelve True si el puerto está abierto."""
try:
with socket.create_connection((ip, puerto), timeout=timeout):
return True
except (ConnectionRefusedError, TimeoutError, OSError):
return False
def escanear_puerto(args):
"""Función que envuelve probar_puerto para usar con executor."""
ip, puerto = args
if probar_puerto(ip, puerto):
with lock:
abiertos.append(puerto)
print(f"[+] {ip}:{puerto} ABIERTO")
ip_objetivo = "192.168.1.1"
puertos = range(1, 1025)
# ThreadPoolExecutor: pool de N hilos que procesan una cola de tareas
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
tareas = [(ip_objetivo, p) for p in puertos]
executor.map(escanear_puerto, tareas)
print(f"\n[*] Puertos abiertos: {sorted(abiertos)}")
asyncio: I/O asíncrono sin hilos
asyncio es más eficiente que threading para operaciones de red porque no crea un hilo por conexión — un solo hilo gestiona miles de conexiones de forma concurrente usando el bucle de eventos del sistema operativo. El resultado es menos memoria y menos overhead de context switching.
import asyncio
async def probar_puerto(ip, puerto, timeout=1):
"""Intenta abrir una conexión TCP. Devuelve el puerto si está abierto."""
try:
_, writer = await asyncio.wait_for(
asyncio.open_connection(ip, puerto),
timeout=timeout
)
writer.close()
await writer.wait_closed()
return puerto
except (ConnectionRefusedError, TimeoutError, OSError, asyncio.TimeoutError):
return None
async def escanear(ip, puertos, concurrencia=500):
"""Escanea una lista de puertos con un máximo de N conexiones simultáneas."""
semaforo = asyncio.Semaphore(concurrencia) # límite de conexiones a la vez
abiertos = []
async def con_semaforo(puerto):
async with semaforo:
return await probar_puerto(ip, puerto)
tareas = [con_semaforo(p) for p in puertos]
resultados = await asyncio.gather(*tareas)
abiertos = [p for p in resultados if p is not None]
return sorted(abiertos)
async def main():
ip = "192.168.1.1"
puertos = range(1, 65536)
print(f"[*] Escaneando {ip} ({len(list(puertos))} puertos)...")
abiertos = await escanear(ip, puertos, concurrencia=500)
for p in abiertos:
print(f"[+] Puerto {p}/tcp ABIERTO")
print(f"\n[*] Total: {len(abiertos)} puertos abiertos")
asyncio.run(main())
Cuándo usar cada uno
"""
threading / ThreadPoolExecutor:
✓ Más sencillo de escribir e integrar con código síncrono
✓ Compatible con librerías que no soportan asyncio (requests, subprocess...)
✗ Cada hilo consume ~8MB de stack — con 1000 hilos son 8GB
✗ Context switching tiene overhead real
asyncio:
✓ Un solo hilo gestiona miles de conexiones simultáneas
✓ Muy poco overhead de memoria
✗ Requiere que TODA la cadena de llamadas sea async
✗ No funciona con librerías bloqueantes (requests, subprocess sin adaptador)
Regla práctica para escaneos:
→ Puerto/host discovery (miles de conexiones cortas): asyncio
→ Llamadas a subprocess o herramientas externas: ThreadPoolExecutor
→ Mezcla de ambas: ThreadPoolExecutor con loop.run_in_executor
"""
import asyncio
import subprocess
import concurrent.futures
async def escanear_con_nmap(ip, executor):
"""Lanzar nmap (bloqueante) desde asyncio usando un executor."""
loop = asyncio.get_event_loop()
resultado = await loop.run_in_executor(
executor,
lambda: subprocess.run(
["nmap", "-sV", "--open", ip],
capture_output=True, text=True
)
)
return resultado.stdout
async def main():
ips = ["192.168.1.1", "192.168.1.2", "192.168.1.3"]
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
tareas = [escanear_con_nmap(ip, executor) for ip in ips]
resultados = await asyncio.gather(*tareas)
for ip, salida in zip(ips, resultados):
print(f"--- {ip} ---\n{salida}")
3.4.3 Manejo de excepciones en scripts ofensivos
Un script que falla sin mensaje de error a las 3 AM cuando estás en medio de un pentest es peor que no tener el script. El manejo de excepciones en scripts ofensivos tiene requisitos específicos: el error de una conexión no debe parar el escaneo completo, los timeouts deben ser agresivos, y el resultado de cada intento — éxito o fracaso — tiene que quedar registrado de forma útil.
Excepciones comunes en operaciones de red
import socket
import requests
import subprocess
# Jerarquía de excepciones de socket (de más específica a más general)
# ConnectionRefusedError → puerto cerrado activamente
# TimeoutError → sin respuesta en el tiempo dado
# socket.gaierror → fallo de resolución DNS
# ConnectionResetError → conexión reiniciada por el remoto
# OSError → clase padre de todos los anteriores
def conectar(ip, puerto, timeout=2):
try:
with socket.create_connection((ip, puerto), timeout=timeout) as s:
s.settimeout(timeout)
banner = s.recv(1024).decode("utf-8", errors="replace").strip()
return {"estado": "abierto", "banner": banner}
except ConnectionRefusedError:
return {"estado": "cerrado", "error": "connection refused"}
except TimeoutError:
return {"estado": "filtrado", "error": "timeout"}
except socket.gaierror as e:
return {"estado": "error", "error": f"DNS: {e}"}
except OSError as e:
return {"estado": "error", "error": str(e)}
# requests: excepciones más específicas
def get_seguro(url, proxies=None, timeout=10):
try:
r = requests.get(url, proxies=proxies, timeout=timeout,
allow_redirects=True, verify=False)
r.raise_for_status() # lanza HTTPError para códigos 4xx/5xx
return r
except requests.exceptions.ConnectionError:
print(f"[-] Sin conexión: {url}")
except requests.exceptions.Timeout:
print(f"[-] Timeout: {url}")
except requests.exceptions.HTTPError as e:
print(f"[-] HTTP {e.response.status_code}: {url}")
except requests.exceptions.RequestException as e:
print(f"[-] Error: {e}")
return None
Logging estructurado para scripts ofensivos
import logging
import sys
from datetime import datetime
def configurar_log(archivo=None, verbose=False):
"""
Configura logging para mostrar en terminal Y guardar en archivo.
Terminal: solo INFO y superior.
Archivo: todo incluyendo DEBUG.
"""
nivel_terminal = logging.DEBUG if verbose else logging.INFO
fmt = "%(asctime)s [%(levelname)s] %(message)s"
datefmt = "%H:%M:%S"
logger = logging.getLogger("pentest")
logger.setLevel(logging.DEBUG)
# Handler de terminal con colores simples
ch = logging.StreamHandler(sys.stdout)
ch.setLevel(nivel_terminal)
ch.setFormatter(logging.Formatter(fmt, datefmt))
logger.addHandler(ch)
# Handler de archivo (guarda todo)
if archivo:
fh = logging.FileHandler(archivo)
fh.setLevel(logging.DEBUG)
fh.setFormatter(logging.Formatter(fmt, datefmt))
logger.addHandler(fh)
return logger
log = configurar_log(
archivo=f"scan_{datetime.now():%Y%m%d_%H%M%S}.log",
verbose=True
)
# Uso en el script
log.info("Iniciando escaneo de 192.168.1.0/24")
log.debug("Timeout configurado: 2s")
log.warning("Host 192.168.1.5 responde lento")
log.error("Fallo al conectar con 192.168.1.10:22")
log.critical("Interrupción del usuario")
Patrones de robustez en scripts largos
import time
import signal
import sys
# Manejar Ctrl+C limpiamente (guardar resultados antes de salir)
resultados = []
def salida_limpia(sig, frame):
print(f"\n[!] Interrumpido. {len(resultados)} resultados guardados.")
guardar_resultados(resultados)
sys.exit(0)
signal.signal(signal.SIGINT, salida_limpia)
# Reintentar operaciones que pueden fallar transitoriamente
def con_reintentos(func, args=(), kwargs={}, intentos=3, espera=1):
for i in range(intentos):
try:
return func(*args, **kwargs)
except Exception as e:
if i == intentos - 1:
raise # último intento: dejar que falle
time.sleep(espera * (i + 1)) # espera progresiva
# Context manager para medir tiempo de operaciones
from contextlib import contextmanager
@contextmanager
def cronometrar(nombre):
inicio = time.time()
try:
yield
finally:
duracion = time.time() - inicio
log.debug(f"{nombre} completado en {duracion:.2f}s")
# Uso:
with cronometrar("Escaneo de puertos"):
abiertos = escanear(ip, puertos)
# Nunca capturar Exception en silencio en un bucle crítico
for ip in lista_ips:
try:
resultado = procesar(ip)
resultados.append(resultado)
except KeyboardInterrupt:
raise # dejar que el handler de SIGINT actúe
except Exception as e:
log.error(f"Error procesando {ip}: {e}")
# continuar con la siguiente IP, no abortar todo
3.4.4 Scapy: crafting de paquetes desde Python
Scapy es una librería Python que permite construir, enviar, capturar y analizar paquetes de red a cualquier nivel del stack. A diferencia de nmap o hping3, en Scapy controlas cada campo de cada cabecera manualmente. Es la herramienta de referencia cuando necesitas hacer algo que ninguna herramienta existente hace exactamente como quieres.
sudo python3 script.py o dar la capability CAP_NET_RAW al intérprete. En Windows requiere WinPcap o Npcap instalado.
Construcción de paquetes: el operador /
from scapy.all import *
# El operador / apila capas (como capas en un paquete real)
# Ethernet / IP / TCP / payload
# Paquete ICMP ping
paquete = IP(dst="192.168.1.1") / ICMP()
respuesta = sr1(paquete, timeout=2, verbose=0)
if respuesta:
print(f"Host activo: {respuesta.src}")
# Ver todos los campos de una capa
IP().show()
TCP().show()
# Paquete TCP SYN manualmente
syn = IP(dst="192.168.1.1") / TCP(dport=80, flags="S")
# flags: S=SYN, A=ACK, F=FIN, R=RST, P=PSH, U=URG
# Enviar y recibir respuesta
# sr1: send and receive one (espera una sola respuesta)
# sr: send and receive (espera múltiples respuestas)
# send: enviar sin esperar respuesta
respuesta = sr1(syn, timeout=2, verbose=0)
if respuesta and respuesta.haslayer(TCP):
flags = respuesta[TCP].flags
if flags == "SA": # SYN-ACK: puerto abierto
print(f"Puerto 80 ABIERTO")
# Enviar RST para cerrar la conexión limpiamente
send(IP(dst="192.168.1.1") / TCP(dport=80, flags="R"), verbose=0)
elif flags == "RA": # RST-ACK: puerto cerrado
print(f"Puerto 80 CERRADO")
ARP ping y SYN scanner
from scapy.all import ARP, Ether, IP, TCP, srp, sr
# ARP ping: descubrir hosts en la red local
# (más fiable que ICMP en redes locales donde el ping puede estar bloqueado)
def arp_ping(red="192.168.1.0/24"):
paquete = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(pdst=red)
respondieron, _ = srp(paquete, timeout=2, verbose=0)
hosts = []
for enviado, recibido in respondieron:
hosts.append({"ip": recibido.psrc, "mac": recibido.hwsrc})
print(f" {recibido.psrc:16s} {recibido.hwsrc}")
return hosts
print("[*] Hosts activos en la red:")
hosts = arp_ping("192.168.1.0/24")
# SYN scanner: más sigiloso que un connect scan
# No completa el three-way handshake → no aparece en logs de aplicación
def syn_scan(ip, puertos):
abiertos = []
paquetes = [IP(dst=ip) / TCP(dport=p, flags="S") for p in puertos]
respondieron, _ = sr(paquetes, timeout=2, verbose=0)
for enviado, recibido in respondieron:
if recibido.haslayer(TCP):
puerto = recibido[TCP].sport
flags = recibido[TCP].flags
if flags == "SA":
abiertos.append(puerto)
send(IP(dst=ip) / TCP(dport=puerto, flags="R"), verbose=0)
return sorted(abiertos)
puertos = list(range(1, 1025))
print(f"\n[*] SYN scan sobre 192.168.1.1...")
abiertos = syn_scan("192.168.1.1", puertos)
print(f"[+] Puertos abiertos: {abiertos}")
Captura y análisis de tráfico
from scapy.all import sniff, IP, TCP, UDP, DNS, DNSQR, wrpcap, rdpcap
# Capturar 50 paquetes en la interfaz eth0
paquetes = sniff(iface="eth0", count=50, filter="tcp port 80")
# Captura con callback por paquete (en tiempo real)
def analizar(pkt):
if pkt.haslayer(DNS) and pkt.haslayer(DNSQR):
dominio = pkt[DNSQR].qname.decode("utf-8", errors="replace")
print(f"[DNS] {pkt[IP].src} → {dominio}")
sniff(iface="eth0", prn=analizar, filter="udp port 53", timeout=30)
# Guardar captura a PCAP
paquetes = sniff(iface="eth0", count=100)
wrpcap("captura.pcap", paquetes)
# Leer un PCAP existente y analizarlo
paquetes = rdpcap("captura.pcap")
for pkt in paquetes:
if pkt.haslayer(TCP) and pkt.haslayer("Raw"):
payload = pkt["Raw"].load
if b"password" in payload.lower():
print(f"[!] Posible credencial: {payload[:100]}")
3.4.5 Requests + BeautifulSoup: scraping y automatización web
Requests es la librería estándar para hacer peticiones HTTP en Python — más ergonómica que urllib y sin el overhead de Selenium. BeautifulSoup parsea el HTML de las respuestas y permite extraer datos con selectores CSS o XPath. Juntas son la base de cualquier herramienta de OSINT web, fuzzer de directorios o automatizador de formularios.
Requests: peticiones HTTP con control total
import requests
# Deshabilitar advertencias de certificados SSL inválidos (en pentesting es habitual)
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Session: reutiliza conexiones y guarda cookies automáticamente
s = requests.Session()
s.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36",
"Accept-Language": "es-ES,es;q=0.9"
})
# GET básico
r = s.get("https://objetivo.com/login", verify=False, timeout=10)
print(r.status_code, r.headers["Content-Type"])
# POST con formulario
datos = {"username": "admin", "password": "password123", "submit": "Login"}
r = s.post("https://objetivo.com/login", data=datos, verify=False)
# La sesión guarda las cookies de autenticación automáticamente
# Comprobar si el login funcionó
if "Dashboard" in r.text or r.url.endswith("/dashboard"):
print("[+] Login exitoso")
# Usar Tor como proxy
proxies_tor = {
"http": "socks5h://127.0.0.1:9050",
"https": "socks5h://127.0.0.1:9050"
}
r = s.get("https://check.torproject.org/api/ip", proxies=proxies_tor, timeout=30)
print(r.json()) # {"IsTor": true, "IP": "x.x.x.x"}
# Subir un archivo
with open("webshell.php", "rb") as f:
r = s.post("https://objetivo.com/upload",
files={"archivo": ("foto.php", f, "image/jpeg")})
# Petición con cabeceras personalizadas (bypass de protecciones)
r = s.get("https://objetivo.com/admin",
headers={"X-Forwarded-For": "127.0.0.1", "X-Real-IP": "127.0.0.1"})
BeautifulSoup: parsear HTML
from bs4 import BeautifulSoup
import requests
r = requests.get("https://objetivo.com", verify=False, timeout=10)
soup = BeautifulSoup(r.text, "html.parser")
# Extraer todos los enlaces
for a in soup.find_all("a", href=True):
print(a["href"])
# Solo enlaces internos (no empiezan por http)
internos = [a["href"] for a in soup.find_all("a", href=True)
if not a["href"].startswith(("http", "//", "#", "mailto"))]
# Extraer texto de elementos por selector CSS
titulo = soup.select_one("h1").get_text(strip=True) if soup.select_one("h1") else ""
emails = soup.find_all(string=lambda t: "@" in str(t) and "." in str(t))
# Extraer campos de formularios (fundamental para automatizar logins)
for form in soup.find_all("form"):
print(f"\nFormulario: action={form.get('action')} method={form.get('method','get')}")
for campo in form.find_all(["input", "textarea", "select"]):
nombre = campo.get("name", "")
tipo = campo.get("type", "text")
valor = campo.get("value", "")
print(f" [{tipo}] {nombre} = {valor!r}")
# Extraer tokens CSRF ocultos
csrf = soup.find("input", {"name": lambda n: n and "csrf" in n.lower()})
if csrf:
token = csrf.get("value", "")
print(f"[*] Token CSRF: {token}")
Fuzzer de directorios con Requests
import requests
import concurrent.futures
import urllib3
urllib3.disable_warnings()
def probar_ruta(base_url, ruta, session):
"""Devuelve (ruta, status_code) si la respuesta no es 404."""
url = f"{base_url.rstrip('/')}/{ruta.strip('/')}"
try:
r = session.get(url, verify=False, timeout=5, allow_redirects=False)
if r.status_code not in (404, 400, 410):
return ruta, r.status_code, len(r.content)
except requests.exceptions.RequestException:
pass
return None
def fuzzear(base_url, wordlist_path, workers=20):
with open(wordlist_path) as f:
palabras = [l.strip() for l in f if l.strip() and not l.startswith("#")]
s = requests.Session()
s.headers["User-Agent"] = "Mozilla/5.0"
encontrados = []
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
futuros = {executor.submit(probar_ruta, base_url, p, s): p for p in palabras}
for futuro in concurrent.futures.as_completed(futuros):
resultado = futuro.result()
if resultado:
ruta, codigo, tamano = resultado
encontrados.append(resultado)
print(f"[{codigo}] {base_url}/{ruta} ({tamano} bytes)")
return sorted(encontrados, key=lambda x: x[1])
resultados = fuzzear(
"https://objetivo.com",
"/usr/share/wordlists/dirbuster/directory-list-2.3-medium.txt",
workers=30
)