"""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()