Semana 15 — Seguridad de redes y cierre del proyecto
Objetivos
- Reconocer ataques comunes a las redes y clasificarlos según la capa que afectan.
- Explicar cómo funcionan los firewalls, IPsec y las VPNs, y qué protege cada uno.
- Configurar un firewall y una VPN con WireGuard.
- Presentar y defender el proyecto.
Sesión A — Seguridad de redes: ataques, firewalls, IPsec y VPNs
Lecturas:
- Kurose y Ross, secs. 1.6 (Networks Under Attack), 8.1 (What Is Network Security?), 8.7 (Network-Layer Security: IPsec and Virtual Private Networks) y 8.9 (Operational Security: Firewalls and Intrusion Detection Systems).
- WireGuard: Next Generation Kernel Network Tunnel, secciones 1 y 2.
Lecturas complementarias:
- Kurose y Ross, sec. 8.8 (Securing Wireless LANs and 4G/5G Cellular Networks).
- RFC 4301, arquitectura de IPsec.
Temas:
- Modelo de amenazas: quién ataca, qué puede ver y qué puede modificar.
- Ataques por capa: ARP spoofing, IP spoofing, DoS y DDoS por amplificación, SYN flood, envenenamiento de DNS y hijack de BGP. Recordar lo visto en semanas anteriores.
- Firewalls: sin estado y con estado. Zonas y DMZ. Sistemas de detección de intrusos.
- Zero trust: no confiar en alguien solo por estar dentro de la red.
- IPsec: AH y ESP, modos túnel y transporte, y security associations.
- WireGuard: un diseño más simple, con claves públicas e identidad por clave.
- Error común: creer que una VPN hace anónimo o seguro a todo. Protege el tramo hasta su otro extremo; lo que pase después ya no lo cubre.
Sesión B — Presentación de proyectos
Lecturas:
- Releer el enunciado de P5.
Temas:
- Cada equipo muestra
wasirespondiendo sobrerimayen Qhapaq Ñan, con un nombre en DNS. - El docente corta un enlace con otro camino disponible mientras se baja un archivo grande, y la transferencia debe terminar correctamente. Las condiciones están en el enunciado de P5.
- Preguntas sobre las decisiones de diseño y las mediciones.
Laboratorio — WireGuard y nftables
Trabajen en code/week-15/.
- Levantar una topología con una red interna, una DMZ y un router que las separa de afuera.
- Con
nftables, escribir un firewall con estado: de adentro hacia afuera sí, de afuera hacia adentro solo el servidor de la DMZ. - Probar cada regla con
ncycurl, y anotar lo que se esperaba frente a lo que pasó. - Crear claves con
wg genkeyy levantar un túnel WireGuard entre dos sitios. - Capturar a cada lado del túnel y comparar el paquete interno con el externo.
- Medir con
iperf3el costo de la VPN en throughput y en MTU.
Entregar P5.
Entregas
- P5 entrega.
Preguntas de repaso
- ¿Por qué un ataque de amplificación usa UDP y direcciones falsificadas? ¿Qué pueden hacer los ISPs para evitarlo?
- Escriban las reglas de un firewall con estado para un servidor web en una DMZ.
- ¿Qué diferencia hay entre el modo túnel y el modo transporte de IPsec? ¿Cuál se usa entre dos sitios?
- ¿Qué protege WireGuard y qué no?
- Para un ataque de cada capa del curso, indiquen qué defensa vieron y qué limitación tiene.
Código de la semana
Compilar y probar dentro del contenedor del curso con make y make test.
Ejemplos (3)Pruebas (1)Laboratorio (1)Compilación (1)
Ejemplos
Los programas que usan los apuntes y la clase.
"""Ayudas de la semana 15: levantar la topología de los dos sitios, correr comandos en sus
nodos, generar claves de WireGuard y elegir el firewall del laboratorio.
"""
import os
import subprocess
import time
import unittest
HERE = os.path.dirname(os.path.abspath(__file__))
CAPTURES = os.path.join(HERE, "captures")
KEYS = os.path.join(HERE, "build", "keys")
# La solución no se publica. Sin ella, se usa el firewall que ustedes escriben en lab/.
ANSWERS = os.path.join(HERE, "solution")
if not os.path.isdir(ANSWERS):
ANSWERS = os.path.join(HERE, "lab")
def answer_path(name):
"""Devuelve la ruta de un archivo del laboratorio, o salta si sigue con el TODO."""
path = os.path.join(ANSWERS, name)
with open(path, encoding="utf-8") as file:
if "TODO" in file.read():
raise unittest.SkipTest(f"falta completar lab/{name}")
return path
class Lab:
"""La topología de Containerlab, que se levanta y se destruye desde las pruebas."""
def __init__(self, folder="sitios", name="w15-sitios", network="redes-w15-sitios"):
self.folder = os.path.join(HERE, folder)
self.topology = f"{folder}.clab.yml"
self.prefix = f"clab-{name}-"
self.network = network
def deploy(self):
done = subprocess.run(
["containerlab", "deploy", "-t", self.topology, "--reconfigure"],
cwd=self.folder, capture_output=True, text=True, timeout=300)
if done.returncode != 0:
raise RuntimeError(done.stdout + done.stderr)
def destroy(self):
subprocess.run(["containerlab", "destroy", "-t", self.topology, "--cleanup"],
cwd=self.folder, capture_output=True, text=True, timeout=300)
subprocess.run(["docker", "network", "rm", self.network], capture_output=True, text=True)
def sh(self, node, *command, timeout=90):
"""Corre un comando en un nodo y devuelve el resultado completo."""
return subprocess.run(["docker", "exec", self.prefix + node, *command],
capture_output=True, text=True, timeout=timeout)
def out(self, node, *command, timeout=90):
"""Corre un comando en un nodo y devuelve stdout y stderr juntos."""
done = self.sh(node, *command, timeout=timeout)
return done.stdout + done.stderr
def shell(self, node, script, timeout=90):
"""Corre una línea de shell (sh -c) en un nodo."""
return subprocess.run(["docker", "exec", self.prefix + node, "sh", "-c", script],
capture_output=True, text=True, timeout=timeout)
def daemon(self, node, command):
"""Deja un comando corriendo en segundo plano dentro de un nodo."""
subprocess.run(["docker", "exec", "-d", self.prefix + node, "sh", "-c", command],
capture_output=True, text=True, check=True)
def load_nft(self, node, path):
"""Carga un archivo de reglas de nftables en un nodo, desde el host."""
with open(path, encoding="utf-8") as file:
done = subprocess.run(["docker", "exec", "-i", self.prefix + node, "nft", "-f", "-"],
input=file.read(), capture_output=True, text=True, timeout=30)
if done.returncode != 0:
raise RuntimeError(done.stdout + done.stderr)
def gen_keys():
"""Crea las claves de WireGuard de fw y gw2 en build/keys, con permisos restringidos."""
os.makedirs(KEYS, exist_ok=True)
os.chmod(KEYS, 0o700)
for who in ("fw", "gw2"):
key = os.path.join(KEYS, f"{who}.key")
pub = os.path.join(KEYS, f"{who}.pub")
private = subprocess.run(["wg", "genkey"], capture_output=True, text=True, check=True).stdout.strip()
public = subprocess.run(["wg", "pubkey"], input=private, capture_output=True, text=True, check=True).stdout.strip()
old = os.umask(0o077)
try:
with open(key, "w", encoding="utf-8") as file:
file.write(private + "\n")
finally:
os.umask(old)
with open(pub, "w", encoding="utf-8") as file:
file.write(public + "\n")
return KEYS
def read_key(name):
with open(os.path.join(KEYS, name), encoding="utf-8") as file:
return file.read().strip()
def setup_tunnel(lab):
"""Levanta el túnel WireGuard entre fw y gw2 con las claves de build/keys.
La tabla de cryptokey routing: cada peer solo acepta las direcciones de la lista allowed-ips.
En fw, el peer gw2 trae la red de b (10.215.3.0/24); en gw2, el peer fw trae la oficina.
"""
fw_priv, fw_pub = read_key("fw.key"), read_key("fw.pub")
gw2_priv, gw2_pub = read_key("gw2.key"), read_key("gw2.pub")
lab.shell("fw", f"""
set -e
printf '%s' '{fw_priv}' > /tmp/wg.key
ip link add wg0 type wireguard
wg set wg0 private-key /tmp/wg.key listen-port 51820
rm -f /tmp/wg.key
wg set wg0 peer '{gw2_pub}' endpoint 100.64.15.70:51820 \
allowed-ips 10.215.3.0/24,10.215.9.2/32 persistent-keepalive 25
ip addr add 10.215.9.1/30 dev wg0
ip link set wg0 up
ip route add 10.215.3.0/24 dev wg0
""")
lab.shell("gw2", f"""
set -e
printf '%s' '{gw2_priv}' > /tmp/wg.key
ip link add wg0 type wireguard
wg set wg0 private-key /tmp/wg.key listen-port 51820
rm -f /tmp/wg.key
wg set wg0 peer '{fw_pub}' endpoint 100.64.15.66:51820 \
allowed-ips 10.215.1.0/24,10.215.9.1/32 persistent-keepalive 25
ip addr add 10.215.9.2/30 dev wg0
ip link set wg0 up
ip route add 10.215.1.0/24 dev wg0
""")
# Si el firewall está cargado (tabla inet fw), hay que dejar pasar el tráfico de la oficina
# hacia el túnel y de vuelta; sin firewall el forwarding ya está abierto.
if lab.shell("fw", "nft list table inet fw").returncode == 0:
lab.shell("fw", "nft add rule inet fw forward iifname eth1 oifname wg0 accept; "
"nft add rule inet fw forward iifname wg0 oifname eth1 accept")
return fw_pub, gw2_pub
def wait_until(predicate, seconds=30, step=0.25):
"""Espera a que una condición se cumpla; devuelve si se cumplió."""
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
if predicate():
return True
time.sleep(step)
return predicate()"""Los experimentos de la semana: firewall con estado, túnel WireGuard e IPsec.
python3 scenarios.py lista los escenarios
python3 scenarios.py firewall corre uno
Cada escenario imprime los comandos que corre: léanlos, porque son lo mismo que harían a mano
en el laboratorio. Hace falta la topología levantada (make up) y las claves (make).
"""
import os
import subprocess
import sys
import time
from netlab import CAPTURES, Lab, setup_tunnel, wait_until
LAB = Lab()
HERE = os.path.dirname(os.path.abspath(__file__))
VERBOSE = True
def say(node, command):
if VERBOSE:
print(f" {node}# {command}")
def report(text):
if VERBOSE:
print(text.rstrip())
def run(node, command):
"""Corre un comando visible en un nodo; falla si el comando falla."""
say(node, command)
done = LAB.shell(node, command)
if done.returncode != 0:
raise RuntimeError(f"{node}: {command}\n{done.stdout}{done.stderr}")
return done.stdout + done.stderr
def quiet(node, command):
return LAB.shell(node, command)
def serve(node, addr, port=80):
"""Deja un servidor web mínimo en un nodo, sobre una dirección y un puerto."""
LAB.daemon(node, f"cd /tmp && printf 'ok\\n' > index.html && "
f"python3 -m http.server {port} --bind {addr}")
wait_until(lambda: f":{port} " in LAB.out(node, "ss", "-tln"), seconds=5)
def firewall_path():
"""El firewall que se va a cargar: la solución si está, si no el del laboratorio."""
for folder in ("solution", "lab"):
path = os.path.join(HERE, folder, "firewall.nft")
if os.path.isfile(path):
with open(path, encoding="utf-8") as file:
if "TODO" in file.read():
continue
return path
print("Falta completar lab/firewall.nft (todavía tiene TODO).")
sys.exit(1)
def load_firewall():
path = firewall_path()
say("fw", f"nft -f {os.path.basename(path)}")
with open(path, encoding="utf-8") as file:
done = subprocess.run(["docker", "exec", "-i", LAB.prefix + "fw", "nft", "-f", "-"],
input=file.read(), capture_output=True, text=True)
if done.returncode != 0:
raise RuntimeError(done.stdout + done.stderr)
def timed_probe(node, host, port, label):
"""Intenta conectar con nc y muestra lo medido: si conectó y cuánto tardó."""
start = time.monotonic()
rc = LAB.shell(node, f"nc -w3 -z {host} {port}").returncode
elapsed = time.monotonic() - start
report(f" {label}: {'conecta' if rc == 0 else 'no conecta'}, {elapsed:.2f} s")
def probe(node, url, label):
"""Pide una URL con curl y muestra el código HTTP, o el error y cuánto tardó en fallar."""
done = LAB.shell(node, f"curl -s -m4 -o /dev/null -w '%{{http_code}} %{{time_total}}' {url}")
code, seconds = done.stdout.split()
report(f" {label}: " + (code if code != "000" else f"falla en {float(seconds):.2f} s"))
return code
def ensure_tunnel():
"""Levanta el túnel si todavía no está, y espera el primer handshake."""
if LAB.shell("fw", "ip link show wg0").returncode != 0:
setup_tunnel(LAB)
quiet("ofi", "ping -c1 -W2 10.215.3.10")
wait_until(lambda: LAB.shell("fw", "wg show wg0 latest-handshakes").stdout.split("\t")[-1].strip()
not in ("", "0"), seconds=10)
def capture(node, iface, name, expression):
"""Empieza una captura en un nodo; devuelve la función que la detiene y la lee."""
path = f"/captures/{name}.pcap"
LAB.daemon(node, f"tcpdump -i {iface} -n -U -w {path} {expression}")
time.sleep(1)
def finish(flags="-n"):
time.sleep(1)
quiet(node, "pkill tcpdump")
time.sleep(0.5)
return LAB.sh(node, "tcpdump", flags, "-r", path).stdout
return finish
# ----------------------------------------------------------------------------------------------
def firewall():
"""El firewall con estado: de afuera solo se alcanza el servidor web de la DMZ."""
serve("web", "100.64.15.10", 80)
serve("ofi", "10.215.1.10", 8080)
print("Antes del firewall (fw solo hace forwarding): afuera llega a todo.")
probe("ext", "http://100.64.15.10/", "ext -> web (DMZ)")
probe("ext", "http://10.215.1.10:8080/", "ext -> ofi (red interna)")
print("\nCargamos el firewall:")
load_firewall()
report(run("fw", "nft list chain inet fw forward"))
print("Después del firewall:")
probe("ext", "http://100.64.15.10/", "ext -> web (DMZ) ")
probe("ext", "http://10.215.1.10:8080/", "ext -> ofi (interna)")
probe("ofi", "http://100.64.15.10/", "ofi -> web (interna a DMZ)")
print("\nTres resultados distintos, medidos con el tiempo de cada intento:")
timed_probe("ext", "100.64.15.10", 443, "ext -> web:443 (aceptado por fw, sin servidor)")
timed_probe("ext", "100.64.15.10", 22, "ext -> web:22 (sin regla, drop en forward) ")
timed_probe("ext", "10.215.1.10", 8080, "ext -> ofi:8080 (red interna, reject) ")
def wireguard():
"""El túnel WireGuard entre los dos sitios, y lo que ve quien está en el medio."""
load_firewall()
ensure_tunnel()
report(run("fw", "wg show wg0"))
report(run("fw", "ip link show wg0"))
serve("b", "10.215.3.10", 80)
print("La oficina alcanza la red del otro sitio, que no tiene ruta pública:")
report(run("ofi", "ping -c2 -W2 10.215.3.10"))
print("\nCapturamos el mismo ping en tres puntos:")
LAB.daemon("fw", "tcpdump -i eth1 -n -c2 icmp -w /captures/tunel-interno-a.pcap")
LAB.daemon("wan", "tcpdump -i eth1 -n -c2 udp port 51820 -w /captures/tunel-externo.pcap")
LAB.daemon("b", "tcpdump -i eth1 -n -c2 icmp -w /captures/tunel-interno-b.pcap")
time.sleep(1)
quiet("ofi", "ping -c2 -W2 10.215.3.10")
time.sleep(2)
print(" en fw eth1 (interno): ICMP en claro")
report(LAB.out("fw", "tcpdump", "-nr", "/captures/tunel-interno-a.pcap"))
print(" en wan eth1 (externo): UDP 51820, cifrado")
report(LAB.out("wan", "tcpdump", "-nr", "/captures/tunel-externo.pcap"))
print(" en b eth1 (interno, destino): ICMP en claro otra vez")
report(LAB.out("b", "tcpdump", "-nr", "/captures/tunel-interno-b.pcap"))
def ipsec():
"""IPsec a mano con ip xfrm: ESP cifra, AH solo autentica. Claves manuales, sin IKE."""
a, b = "100.64.15.100", "100.64.15.97" # ext y wan, enlace directo
import binascii
ek = "0x" + binascii.hexlify(os.urandom(20)).decode()
ak = "0x" + binascii.hexlify(os.urandom(32)).decode()
print("ESP en modo transporte (cifra la carga): clave manual, una SA por sentido.")
for node in ("ext", "wan"):
run(node, f"ip xfrm state flush; ip xfrm policy flush")
run(node, f"ip xfrm state add src {a} dst {b} proto esp spi 0x201 mode transport "
f"aead 'rfc4106(gcm(aes))' {ek} 128")
run(node, f"ip xfrm state add src {b} dst {a} proto esp spi 0x202 mode transport "
f"aead 'rfc4106(gcm(aes))' {ek} 128")
for src, dst, d in ((a, b, "out"), (b, a, "in"), (b, a, "out"), (a, b, "in")):
quiet(node, f"ip xfrm policy add src {src} dst {dst} dir {d} tmpl proto esp mode transport")
quiet("ext", f"ping -c1 -W2 {b}") # resuelve ARP
LAB.daemon("wan", "tcpdump -i eth3 -nv -c2 esp -w /captures/ipsec-esp.pcap")
time.sleep(1)
quiet("ext", f"ping -c2 -W2 {b}")
time.sleep(1)
report(LAB.out("wan", "tcpdump", "-nvr", "/captures/ipsec-esp.pcap"))
print("\nAH en modo transporte (autentica, NO cifra): el ICMP se lee, con cabecera AH.")
for node in ("ext", "wan"):
run(node, f"ip xfrm state flush; ip xfrm policy flush")
run(node, f"ip xfrm state add src {a} dst {b} proto ah spi 0x301 mode transport "
f"auth 'hmac(sha256)' {ak}")
run(node, f"ip xfrm state add src {b} dst {a} proto ah spi 0x302 mode transport "
f"auth 'hmac(sha256)' {ak}")
for src, dst, d in ((a, b, "out"), (b, a, "in"), (b, a, "out"), (a, b, "in")):
quiet(node, f"ip xfrm policy add src {src} dst {dst} dir {d} tmpl proto ah mode transport")
LAB.daemon("wan", "tcpdump -i eth3 -nv -c2 -w /captures/ipsec-ah.pcap")
time.sleep(1)
quiet("ext", f"ping -c2 -W2 {b}")
time.sleep(1)
report(LAB.out("wan", "tcpdump", "-nvr", "/captures/ipsec-ah.pcap"))
for node in ("ext", "wan"):
quiet(node, "ip xfrm state flush; ip xfrm policy flush")
def overhead():
"""Cuánto pesa WireGuard: el tamaño del UDP externo para tres tamaños de ping."""
ensure_tunnel()
for data in (56, 68, 69):
finish = capture("wan", "eth1", f"overhead-{data}", "udp port 51820")
quiet("ofi", f"ping -c1 -W2 -s {data} 10.215.3.10")
first = next(line for line in finish().splitlines() if "UDP" in line)
report(f" -s {data}: interno {data + 28}, afuera {first.split(': ', 1)[1]}")
def allowed():
"""Identidad por clave: un origen fuera de allowed-ips llega cifrado y se descarta."""
ensure_tunnel()
report(run("gw2", "wg show wg0 allowed-ips"))
for source in ("100.64.15.66", "10.215.9.1"):
tunnel = capture("gw2", "eth1", "allowed-udp", "udp port 51820")
inner = capture("b", "eth1", "allowed-icmp", "icmp")
result = LAB.out("fw", "ping", "-I", source, "-c2", "-W2", "10.215.3.10")
seen_inner = len(inner().splitlines())
seen_tunnel = len(tunnel().splitlines())
loss = next(line for line in result.splitlines() if "packet loss" in line)
report(f" origen {source}: {loss}")
report(f" UDP del túnel en gw2: {seen_tunnel} paquetes; ICMP en b: {seen_inner} paquetes")
def pmtu():
"""PMTUD: el ICMP que le avisa a la oficina la MTU del túnel, y lo que guarda."""
ensure_tunnel()
quiet("ofi", "ip route flush cache")
finish = capture("ofi", "eth1", "pmtu", "icmp")
LAB.daemon("b", "iperf3 -s -1 -p 5301")
wait_until(lambda: ":5301 " in LAB.out("b", "ss", "-tln"), seconds=5)
quiet("ofi", "iperf3 -c 10.215.3.10 -p 5301 -t 2")
report("\n".join(finish("-nvt").splitlines()[:4]))
report(run("ofi", "ip route get 10.215.3.10"))
def cost():
"""El costo en throughput: el mismo camino de cuatro saltos, sin cifrar y por el túnel."""
ensure_tunnel()
run("fw", "nft flush ruleset")
run("wan", "ip route add 10.215.3.0/24 via 100.64.15.70")
plain = ("ip route replace 10.215.3.0/24 via 100.64.15.65", "ip route replace 10.215.1.0/24 via 100.64.15.69")
tunnel = ("ip route replace 10.215.3.0/24 dev wg0", "ip route replace 10.215.1.0/24 dev wg0")
try:
for _ in range(3):
for name, routes in (("sin cifrar", plain), ("por el túnel", tunnel)):
quiet("fw", routes[0])
quiet("gw2", routes[1])
quiet("ofi", "ip route flush cache")
LAB.daemon("b", "iperf3 -s -1 -p 5302")
wait_until(lambda: ":5302 " in LAB.out("b", "ss", "-tln"), seconds=5)
out = LAB.out("ofi", "iperf3", "-c", "10.215.3.10", "-p", "5302", "-t", "3", "-f", "m", timeout=30)
line = next(row for row in out.splitlines() if "receiver" in row)
report(f" {name:13s} {' '.join(line.split()[6:8])}")
finally:
quiet("wan", "ip route del 10.215.3.0/24")
quiet("fw", tunnel[0])
quiet("gw2", tunnel[1])
load_firewall()
quiet("fw", "nft add rule inet fw forward iifname eth1 oifname wg0 accept; "
"nft add rule inet fw forward iifname wg0 oifname eth1 accept")
SCENARIOS = {
"firewall": firewall,
"wireguard": wireguard,
"overhead": overhead,
"allowed": allowed,
"pmtu": pmtu,
"cost": cost,
"ipsec": ipsec,
}
def main():
os.makedirs(CAPTURES, exist_ok=True)
if len(sys.argv) != 2 or sys.argv[1] not in SCENARIOS:
print("escenarios:")
for name, func in SCENARIOS.items():
print(f" {name:12s} {func.__doc__.splitlines()[0]}")
return
SCENARIOS[sys.argv[1]]()
if __name__ == "__main__":
main()# Dos sitios de una misma organización, separados por una red que no es de ellos.
#
# ofi 10.215.1.10 --- eth1 eth1 --- gw2 --- eth2 --- b 10.215.3.10
# fw --- eth3 --- wan ---
# web 100.64.15.10 -- eth2 |
# ext 100.64.15.100
#
# fw separa tres zonas: la red interna (eth1), la DMZ (eth2) y afuera (eth3). wan es "Internet":
# sabe llegar a la red interna de fw para que quien la proteja sea el firewall y no la falta de
# una ruta, y no sabe llegar a la red de b, que solo se alcanza por el túnel.
# No hay NAT: así las capturas muestran las direcciones reales.
name: w15-sitios
mgmt:
network: redes-w15-sitios
ipv4-subnet: 172.90.15.0/26
ipv6-subnet: 3fff:172:90:15:1::/80
topology:
defaults:
kind: linux
image: redes-router
binds:
- ../captures:/captures
nodes:
ofi:
exec:
- ip route del default
- sysctl -w net.ipv6.conf.all.disable_ipv6=1
- ip addr add 10.215.1.10/24 dev eth1
- ip route add default via 10.215.1.1
web:
exec:
- ip route del default
- sysctl -w net.ipv6.conf.all.disable_ipv6=1
- ip addr add 100.64.15.10/26 dev eth1
- ip route add default via 100.64.15.1
fw:
exec:
- ip route del default
- sysctl -w net.ipv6.conf.all.disable_ipv6=1
- sysctl -w net.ipv4.ip_forward=1
- ip addr add 10.215.1.1/24 dev eth1
- ip addr add 100.64.15.1/26 dev eth2
- ip addr add 100.64.15.66/30 dev eth3
- ip route add default via 100.64.15.65
wan:
exec:
- ip route del default
- sysctl -w net.ipv6.conf.all.disable_ipv6=1
- sysctl -w net.ipv4.ip_forward=1
- ip addr add 100.64.15.65/30 dev eth1
- ip addr add 100.64.15.69/30 dev eth2
- ip addr add 100.64.15.97/27 dev eth3
- ip route add 10.215.1.0/24 via 100.64.15.66
- ip route add 100.64.15.0/26 via 100.64.15.66
gw2:
exec:
- ip route del default
- sysctl -w net.ipv6.conf.all.disable_ipv6=1
- sysctl -w net.ipv4.ip_forward=1
- ip addr add 100.64.15.70/30 dev eth1
- ip addr add 10.215.3.1/24 dev eth2
- ip route add default via 100.64.15.69
b:
exec:
- ip route del default
- sysctl -w net.ipv6.conf.all.disable_ipv6=1
- ip addr add 10.215.3.10/24 dev eth1
- ip route add default via 10.215.3.1
ext:
exec:
- ip route del default
- sysctl -w net.ipv6.conf.all.disable_ipv6=1
- ip addr add 100.64.15.100/27 dev eth1
- ip route add default via 100.64.15.97
# Containerlab usa MTU 9500 por defecto; el costo de la VPN solo se ve con el valor de Ethernet.
links:
- {endpoints: ["ofi:eth1", "fw:eth1"], mtu: 1500}
- {endpoints: ["web:eth1", "fw:eth2"], mtu: 1500}
- {endpoints: ["fw:eth3", "wan:eth1"], mtu: 1500}
- {endpoints: ["wan:eth2", "gw2:eth1"], mtu: 1500}
- {endpoints: ["wan:eth3", "ext:eth1"], mtu: 1500}
- {endpoints: ["gw2:eth2", "b:eth1"], mtu: 1500}Pruebas
Comprueban lo que afirman los apuntes; se ejecutan con make test.
"""Pruebas de la semana 15: firewall con estado, túnel WireGuard e IPsec.
Levantan la topología de los dos sitios una sola vez, cargan el firewall (la solución si está,
si no el del laboratorio, que se salta mientras tenga TODO) y comprueban lo que afirman los
apuntes. Las mediciones usan márgenes generosos: la red es compartida y ruidosa.
"""
import os
import time
import unittest
import netlab
from netlab import Lab, answer_path, gen_keys, setup_tunnel, wait_until
LAB = Lab()
HAS_FIREWALL = False
def chosen_firewall():
"""Ruta del firewall a cargar, o None si solo existe el del laboratorio con TODO."""
for folder in ("solution", "lab"):
path = os.path.join(netlab.HERE, folder, "firewall.nft")
if os.path.isfile(path):
with open(path, encoding="utf-8") as file:
if "TODO" in file.read():
continue
return path
return None
def serve(node, addr, port=80):
LAB.daemon(node, f"cd /tmp && printf 'ok\\n' > index.html && "
f"python3 -m http.server {port} --bind {addr}")
wait_until(lambda: f":{port} " in LAB.out(node, "ss", "-tln"), seconds=6)
def http_code(node, url, timeout=5):
done = LAB.shell(node, f"curl -s -m{timeout - 1} -o /dev/null -w '%{{http_code}}' {url}")
return done.stdout.strip() or "000"
def setUpModule():
global HAS_FIREWALL
gen_keys()
LAB.deploy()
firewall = chosen_firewall()
HAS_FIREWALL = firewall is not None
if HAS_FIREWALL:
LAB.load_nft("fw", firewall)
serve("web", "100.64.15.10", 80)
serve("b", "10.215.3.10", 80)
serve("ofi", "10.215.1.10", 8080)
# Servicios sin regla en el firewall: si una regla sobrara, estas conexiones entrarían.
LAB.daemon("web", "nc -lk -p 22 -s 100.64.15.10")
wait_until(lambda: ":22 " in LAB.out("web", "ss", "-tln"), seconds=5)
def tearDownModule():
LAB.destroy()
class Firewall(unittest.TestCase):
"""El firewall con estado del sitio A."""
def setUp(self):
if not HAS_FIREWALL:
self.skipTest("falta completar lab/firewall.nft")
def test_outside_reaches_the_dmz_web_server(self):
"""De afuera (ext) se llega al servidor web de la DMZ en el puerto 80."""
self.assertEqual(http_code("ext", "http://100.64.15.10/"), "200")
def test_outside_does_not_reach_the_internal_network(self):
"""De afuera no se llega a la red interna, y el rechazo se nota al instante."""
start = time.monotonic()
self.assertEqual(http_code("ext", "http://10.215.1.10:8080/"), "000")
self.assertLess(time.monotonic() - start, 2.5, "reject falla rápido; un drop agota el timeout")
def test_dmz_cannot_open_connections_to_the_internal_network(self):
"""Si comprometen el servidor de la DMZ, el firewall sigue entre él y las oficinas."""
self.assertNotEqual(LAB.shell("web", "nc -w2 -z 10.215.1.10 8080").returncode, 0)
def test_outside_cannot_open_connections_to_the_router(self):
"""La cadena input protege al propio router: desde afuera no se abre nada en él."""
LAB.daemon("fw", "nc -lk -p 2222")
wait_until(lambda: ":2222 " in LAB.out("fw", "ss", "-tln"), seconds=5)
self.assertNotEqual(LAB.shell("ext", "nc -w2 -z 100.64.15.66 2222").returncode, 0)
LAB.shell("fw", "pkill nc")
def test_replies_of_an_outbound_connection_come_back(self):
"""El estado deja volver las respuestas sin una regla de afuera hacia la red interna."""
serve("ext", "100.64.15.100", 9001)
self.assertEqual(http_code("ofi", "http://100.64.15.100:9001/"), "200")
def test_dmz_port_without_a_rule_is_silently_dropped(self):
"""El puerto 22 de la DMZ sí tiene servidor, pero no tiene regla: se descarta en silencio.
Con un servidor escuchando, si faltara la regla la conexión entraría. Como el firewall
descarta (no rechaza), el cliente no recibe nada y espera el timeout: la prueba exige que
falle y que haya tardado (silencio), no que conteste al instante (RST).
"""
start = time.time()
rc = LAB.shell("ext", "nc -w3 -z 100.64.15.10 22").returncode
elapsed = time.time() - start
self.assertNotEqual(rc, 0, "el puerto 22 no debería ser alcanzable desde afuera")
self.assertGreater(elapsed, 1.5, "debería esperar el timeout (silencio), no recibir un RST")
def test_inside_reaches_the_dmz_and_the_outside(self):
"""De la red interna sí se sale: hacia la DMZ y hacia afuera."""
self.assertEqual(http_code("ofi", "http://100.64.15.10/"), "200")
serve("ext", "100.64.15.100", 9000)
self.assertEqual(http_code("ofi", "http://100.64.15.100:9000/"), "200")
def test_default_policy_is_drop(self):
"""Las cadenas input y forward descartan por defecto."""
rules = LAB.out("fw", "nft", "list", "table", "inet", "fw")
self.assertIn("hook input", rules)
self.assertIn("hook forward", rules)
self.assertEqual(rules.count("policy drop;"), 2)
class Tunnel(unittest.TestCase):
"""El túnel WireGuard entre los dos sitios."""
@classmethod
def setUpClass(cls):
setup_tunnel(LAB)
ready = wait_until(
lambda: LAB.shell("fw", "wg show wg0 latest-handshakes").stdout.split("\t")[-1].strip()
not in ("", "0"), seconds=15)
if not ready:
raise unittest.SkipTest("el handshake de WireGuard no subió")
@classmethod
def tearDownClass(cls):
LAB.shell("fw", "ip link del wg0")
LAB.shell("gw2", "ip link del wg0")
def test_office_reaches_the_other_site(self):
"""La oficina alcanza la red del otro sitio, que no tiene ruta pública."""
out = LAB.out("ofi", "ping", "-c3", "-W2", "10.215.3.10")
self.assertIn(" 0% packet loss", out)
def test_the_middle_sees_only_encrypted_udp(self):
"""Quien captura en el medio ve UDP al puerto 51820, no el ICMP interno."""
LAB.daemon("wan", "tcpdump -i eth1 -n -c4 -w /captures/check-ext.pcap")
time.sleep(1)
LAB.shell("ofi", "ping -c3 -W2 10.215.3.10")
time.sleep(2)
dump = LAB.out("wan", "tcpdump", "-nr", "/captures/check-ext.pcap")
self.assertIn("51820", dump)
self.assertNotIn("10.215.3.10", dump)
self.assertNotIn("ICMP echo", dump)
def test_allowed_ips_drops_a_source_outside_the_list(self):
"""Identidad por clave: gw2 descarta un paquete descifrado cuyo origen no está en la lista.
gw2 acepta de fw solo 10.215.1.0/24 y 10.215.9.1/32. Un ping desde fw con origen
100.64.15.66 (su eth3, fuera de la lista) viaja cifrado y llega a gw2, pero se descarta
tras descifrar: no sale nada hacia b. El control, con un origen permitido, sí llega.
"""
LAB.daemon("b", "tcpdump -i eth1 -n icmp -w /captures/ai-b.pcap")
time.sleep(1)
bad = LAB.out("fw", "ping", "-I", "100.64.15.66", "-c2", "-W2", "10.215.3.10")
self.assertIn(" 100% packet loss", bad)
time.sleep(1)
LAB.shell("b", "pkill tcpdump")
time.sleep(1)
reached = LAB.out("b", "tcpdump", "-nr", "/captures/ai-b.pcap")
self.assertNotIn("echo request", reached) # nada salió en b: gw2 lo descartó
# Control: un origen permitido (la dirección de wg0) sí cruza el túnel.
good = LAB.out("fw", "ping", "-I", "10.215.9.1", "-c2", "-W2", "10.215.3.10")
self.assertIn(" 0% packet loss", good)
def test_pmtud_caches_the_tunnel_mtu(self):
"""La MTU de wg0 es menor que 1500; PMTUD hace que la oficina la cachee."""
mtu = LAB.out("fw", "cat", "/sys/class/net/wg0/mtu").strip()
self.assertLess(int(mtu), 1500)
# Una transferencia grande dispara PMTUD en la ruta hacia b.
LAB.daemon("b", "iperf3 -s -1 -p 5201")
wait_until(lambda: ":5201 " in LAB.out("b", "ss", "-tln"), seconds=5)
LAB.shell("ofi", "iperf3 -c 10.215.3.10 -p 5201 -t 2", timeout=20)
route = LAB.out("ofi", "ip", "route", "get", "10.215.3.10")
self.assertIn(f"mtu {mtu}", route)
def test_padding_makes_the_outer_packet_jump_by_16(self):
"""El UDP externo es el interno redondeado a múltiplo de 16, más 32 bytes."""
for data, expected in ((68, 128), (69, 144)):
LAB.daemon("wan", f"tcpdump -i eth1 -n -U -w /captures/pad{data}.pcap udp port 51820")
time.sleep(1)
LAB.shell("ofi", f"ping -c1 -W2 -s {data} 10.215.3.10")
time.sleep(1)
LAB.shell("wan", "pkill tcpdump")
time.sleep(0.5)
dump = LAB.out("wan", "tcpdump", "-nr", f"/captures/pad{data}.pcap")
self.assertIn(f"100.64.15.70.51820: UDP, length {expected}", dump)
def test_vpn_costs_throughput_but_still_works(self):
"""El túnel tiene un costo, pero una transferencia por él funciona (margen amplio)."""
LAB.daemon("b", "iperf3 -s -1 -p 5202")
wait_until(lambda: ":5202 " in LAB.out("b", "ss", "-tln"), seconds=5)
out = LAB.out("ofi", "iperf3", "-c", "10.215.3.10", "-p", "5202", "-t", "3", timeout=20)
self.assertIn("receiver", out)
self.assertNotIn("connect failed", out)
class IPsec(unittest.TestCase):
"""IPsec a mano con ip xfrm: ESP cifra, AH autentica sin cifrar."""
A = "100.64.15.100" # ext
B = "100.64.15.97" # wan
def tearDown(self):
for node in ("ext", "wan"):
LAB.shell(node, "ip xfrm state flush; ip xfrm policy flush")
def _add(self, proto, args):
a, b = self.A, self.B
for node in ("ext", "wan"):
LAB.shell(node, "ip xfrm state flush; ip xfrm policy flush")
LAB.shell(node, f"ip xfrm state add src {a} dst {b} proto {proto} spi 0x201 {args}")
LAB.shell(node, f"ip xfrm state add src {b} dst {a} proto {proto} spi 0x202 {args}")
for src, dst, d in ((a, b, "out"), (b, a, "in"), (b, a, "out"), (a, b, "in")):
LAB.shell(node, f"ip xfrm policy add src {src} dst {dst} dir {d} "
f"tmpl proto {proto} mode transport")
def test_esp_hides_the_payload(self):
"""Con ESP, en el medio se ve proto ESP y no el ICMP interno."""
self._add("esp", "mode transport aead 'rfc4106(gcm(aes))' "
"0x0102030405060708090a0b0c0d0e0f1011121314 128")
LAB.shell("ext", f"ping -c1 -W2 {self.B}") # resuelve ARP
LAB.daemon("wan", "tcpdump -i eth3 -nv -c2 esp -w /captures/check-esp.pcap")
time.sleep(1)
LAB.shell("ext", f"ping -c2 -W2 {self.B}")
time.sleep(1)
dump = LAB.out("wan", "tcpdump", "-nvr", "/captures/check-esp.pcap")
self.assertIn("ESP", dump)
self.assertNotIn("ICMP echo", dump)
def test_ah_authenticates_but_does_not_hide(self):
"""Con AH, en el medio se ve la cabecera AH y el ICMP sigue legible."""
self._add("ah", "mode transport auth 'hmac(sha256)' "
"0x0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20")
LAB.shell("ext", f"ping -c1 -W2 {self.B}")
LAB.daemon("wan", "tcpdump -i eth3 -nv -c2 -w /captures/check-ah.pcap")
time.sleep(1)
LAB.shell("ext", f"ping -c2 -W2 {self.B}")
time.sleep(1)
dump = LAB.out("wan", "tcpdump", "-nvr", "/captures/check-ah.pcap")
self.assertIn("AH", dump)
self.assertIn("ICMP echo", dump)
class LabTODO(unittest.TestCase):
"""Marca el estado del laboratorio: pasa cuando el firewall está completo."""
def test_firewall_is_written(self):
answer_path("firewall.nft") # salta con un mensaje si sigue con TODO
self.assertTrue(HAS_FIREWALL)
if __name__ == "__main__":
unittest.main()Laboratorio
Archivos de partida del laboratorio de la semana.
#!/usr/sbin/nft -f
# Laboratorio: completen este firewall con estado para fw. Quiten cada TODO y escriban la regla.
# fw separa tres zonas:
# eth1 red interna 10.215.1.0/24
# eth2 DMZ 100.64.15.0/26, con el servidor web en 100.64.15.10
# eth3 afuera
# Objetivo: de adentro hacia afuera sí; de afuera hacia adentro solo el servidor web de la DMZ,
# en los puertos 80 y 443. Prueben cada regla con nc y curl (ver los apuntes).
flush ruleset
table inet fw {
chain input {
type filter hook input priority filter; policy drop;
# TODO: aceptar lo que ya es parte de una conexión conocida, y descartar lo inválido.
iif "lo" accept
iifname "eth1" icmp type echo-request accept
iifname "eth1" tcp dport 22 accept
# TODO: aceptar el UDP del túnel WireGuard (puerto 51820) que llega por eth3.
}
chain forward {
type filter hook forward priority filter; policy drop;
# TODO: aceptar lo establecido y descartar lo inválido (como en input).
# TODO: de eth1 hacia eth3, y de eth2 hacia eth3: aceptar.
# TODO: de eth1 hacia eth2: aceptar.
# TODO: de eth3 hacia la DMZ (100.64.15.10), puertos 80 y 443: aceptar.
# TODO: de eth3 hacia eth1: rechazar con icmp admin-prohibited.
}
chain output {
type filter hook output priority filter; policy accept;
}
}Compilación
Cómo se compilan y ejecutan los ejemplos.
# Semana 15: seguridad de redes. Firewall con estado (nftables), túnel WireGuard e IPsec.
# Se usa dentro del contenedor del curso.
#
# make crea las claves de WireGuard en build/keys y la carpeta de capturas
# make test crea las claves, levanta la topología, corre las pruebas (test_*.py) y
# siempre la destruye. Las pruebas del firewall se saltan mientras
# lab/firewall.nft tenga el TODO (vista del estudiante)
# make up levanta la topología de los dos sitios (sitios/sitios.clab.yml)
# make down destruye la topología y su red de Docker
# make clean lo anterior, y borra claves, capturas y estado de Containerlab
.PHONY: all test up down clean
all: build/keys captures
build/keys:
mkdir -p build
python3 -c "import netlab; netlab.gen_keys()"
captures:
mkdir -p captures
test: all
@status=0; PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -p 'test_*.py' -v || status=$$?; \
$(MAKE) --no-print-directory down; exit $$status
up: all
cd sitios && containerlab deploy -t sitios.clab.yml --reconfigure
down:
@cd sitios && containerlab destroy -t sitios.clab.yml --cleanup >/dev/null 2>&1 || true
@docker network rm redes-w15-sitios >/dev/null 2>&1 || true
clean: down
rm -rf build captures __pycache__ sitios/clab-w15-sitios