"""Pruebas de forwarding, ICMP, fragmentación, colas e IPv6 en fwd/fwd.clab.yml.""" import os import re import subprocess import unittest from netlab import HERE, Lab, answer_path, wait_until LAB = Lab("fwd", "w04-fwd", "redes-w04-fwd") NO_DF_SENDER = """ import socket s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.setsockopt(socket.IPPROTO_IP, 10, 0) # IP_MTU_DISCOVER = IP_PMTUDISC_DONT: sin DF s.sendto(b"x" * 1400, ("10.200.3.10", 9999)) """ def setUpModule(): unittest.addModuleCleanup(LAB.destroy) LAB.deploy() ok = wait_until(lambda: "10.200.3.0/24" in LAB.out("r1", "ip", "route") and "10.200.1.0/24" in LAB.out("r3", "ip", "route")) if not ok: raise RuntimeError("las rutas iniciales no aparecieron") def average_rtt(output): match = re.search(r"min/avg/max = [\d.]+/([\d.]+)/", output) return float(match.group(1)) if match else None def set_link_mtu(mtu): LAB.sh("r1", "ip", "link", "set", "eth3", "mtu", str(mtu)) LAB.sh("r3", "ip", "link", "set", "eth3", "mtu", str(mtu)) class A_StartingState(unittest.TestCase): def test_h1_reaches_h3(self): self.assertEqual(LAB.sh("h1", "ping", "-c", "2", "-W", "2", "10.200.3.10").returncode, 0) def test_no_default_route_through_management(self): self.assertNotIn("default", LAB.out("r1", "ip", "route")) self.assertNotIn("default", LAB.out("r2", "ip", "route")) def test_links_use_ethernet_mtu(self): self.assertIn("mtu 1500", LAB.out("h1", "ip", "link", "show", "eth1")) def test_missing_route_gives_net_unreachable(self): capture = LAB.capture("h1", "eth1", "icmp", 2) LAB.sh("h1", "ping", "-c", "1", "-W", "1", "10.200.2.10") self.assertIn("net 10.200.2.10 unreachable", capture.result()) self.assertIn("unreachable", LAB.out("r1", "ip", "route", "get", "10.200.2.10")) class B_PerHop(unittest.TestCase): """Lo que cambia y lo que no en cada salto, en el camino h1 - r1 - r3 - h3.""" def captures(self): points = [("h1", "eth1"), ("r1", "eth3"), ("r3", "eth1")] running = [LAB.capture(n, i, "icmp[icmptype] == icmp-echo", 1) for n, i in points] LAB.sh("h1", "ping", "-c", "1", "-W", "2", "10.200.3.10") return [c.result() for c in running] def test_ttl_macs_and_ips(self): outputs = self.captures() self.assertEqual([int(re.search(r"ttl (\d+)", o).group(1)) for o in outputs], [64, 63, 62]) for output in outputs: self.assertIn("10.200.1.10 > 10.200.3.10", output) macs = [re.match(r"\S+ (\S+) > (\S+),", o).groups() for o in outputs] self.assertEqual(len(set(macs)), 3) def test_checksum_is_redone_at_each_hop(self): points = [("r1", "eth1"), ("r1", "eth3")] running = [LAB.capture(n, i, "icmp[icmptype] == icmp-echo", 1, flags=("-n", "-x")) for n, i in points] LAB.sh("h1", "ping", "-c", "1", "-W", "2", "10.200.3.10") words = [re.search(r"0x0000:\s+(.*)", c.result()).group(1).split() for c in running] # Palabras 4 (TTL y protocolo) y 5 (checksum) del encabezado. self.assertEqual(int(words[0][4], 16) - int(words[1][4], 16), 0x100) self.assertEqual(int(words[1][5], 16) - int(words[0][5], 16), 0x100) self.assertEqual(words[0][:4], words[1][:4]) class C_Icmp(unittest.TestCase): def test_ttl_exceeded(self): capture = LAB.capture("h1", "eth1", "icmp", 2) LAB.sh("h1", "ping", "-c", "1", "-W", "1", "-t", "1", "10.200.3.10") self.assertIn("10.200.1.1 > 10.200.1.10: ICMP time exceeded", capture.result()) def test_host_unreachable_comes_from_last_router(self): capture = LAB.capture("h1", "eth1", "icmp", 2, seconds=9) LAB.sh("h1", "ping", "-c", "1", "-W", "6", "10.200.3.99") self.assertIn("10.200.13.2 > 10.200.1.10: ICMP host 10.200.3.99 unreachable", capture.result()) def test_port_unreachable_comes_from_destination(self): capture = LAB.capture("h1", "eth1", "icmp", 1) LAB.sh("h1", "sh", "-c", "echo hola | nc -u -w1 10.200.3.10 9999") self.assertIn("10.200.3.10 udp port 9999 unreachable", capture.result()) def test_traceroute_ends_at_destination(self): lines = LAB.out("h1", "traceroute", "-n", "-w", "1", "10.200.3.10").splitlines() self.assertIn("10.200.1.1", lines[1]) self.assertIn("10.200.13.2", lines[2]) self.assertIn("10.200.3.10", lines[3]) class D_Fragmentation(unittest.TestCase): def setUp(self): set_link_mtu(600) LAB.sh("h1", "ip", "route", "flush", "cache") self.addCleanup(LAB.sh, "h1", "ip", "route", "flush", "cache") self.addCleanup(set_link_mtu, 1500) def test_df_gets_need_to_frag_and_host_then_fragments(self): capture = LAB.capture("h1", "eth1", "icmp", 2) first = LAB.sh("h1", "ping", "-c", "1", "-W", "1", "-s", "1400", "10.200.3.10") self.assertNotEqual(first.returncode, 0) output = capture.result() self.assertIn("flags [DF]", output) self.assertIn("need to frag (mtu 600)", output) self.assertIn("mtu 600", LAB.out("h1", "ip", "route", "get", "10.200.3.10")) # h1 ya conoce el MTU del camino: fragmenta él, y el ping funciona. at_source = LAB.capture("h1", "eth1", "ip[6:2] & 0x3fff != 0 and src 10.200.1.10", 3) second = LAB.sh("h1", "ping", "-c", "1", "-W", "2", "-s", "1400", "10.200.3.10") self.assertEqual(second.returncode, 0) self.assertEqual(len(re.findall(r"offset \d+", at_source.result())), 3) def test_without_df_the_router_fragments(self): before = LAB.capture("h1", "eth1", "udp", 1) after = LAB.capture("r1", "eth3", "src 10.200.1.10", 3) self.assertEqual(LAB.python("h1", NO_DF_SENDER), "") self.assertIn("offset 0, flags [none]", before.result()) offsets = re.findall(r"offset (\d+), flags \[(\+|none)\]", after.result()) self.assertEqual(offsets, [("0", "+"), ("576", "+"), ("1152", "none")]) class E_Queues(unittest.TestCase): def setUp(self): LAB.sh("r1", "tc", "qdisc", "add", "dev", "eth3", "root", "handle", "1:", "tbf", "rate", "10mbit", "burst", "16kb", "limit", "256kb") self.addCleanup(LAB.sh, "r1", "tc", "qdisc", "del", "dev", "eth3", "root") LAB.daemon("h3", "iperf3 -s -1") wait_until(lambda: "5201" in LAB.out("h3", "netstat", "-tln"), seconds=5) def load(self): return subprocess.Popen( ["docker", "exec", LAB.prefix + "h1", "iperf3", "-c", "10.200.3.10", "-t", "7"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def test_queue_delay_and_strict_priority(self): idle = average_rtt(LAB.out("h1", "ping", "-c", "5", "-i", "0.2", "10.200.3.10")) traffic = self.load() self.addCleanup(traffic.wait) wait_until(lambda: "backlog 0b" not in LAB.out("r1", "tc", "-s", "qdisc", "show", "dev", "eth3"), seconds=4, step=0.2) loaded = average_rtt(LAB.out("h1", "ping", "-c", "8", "-i", "0.25", "10.200.3.10")) # Las colas dependen de la máquina: solo se exige que el retardo se note. self.assertGreater(loaded, 10) self.assertGreater(loaded, 5 * idle) traffic.wait() # Con una banda de prioridad para ICMP, el ping se adelanta a la cola. LAB.sh("r1", "tc", "qdisc", "add", "dev", "eth3", "parent", "1:1", "handle", "10:", "prio") for rule in ("add table ip qos", "add chain ip qos mark_icmp { type filter hook forward priority 0; }", "add rule ip qos mark_icmp ip protocol icmp meta priority set 10:1"): self.assertEqual(LAB.sh("r1", "nft", rule).returncode, 0, rule) self.addCleanup(LAB.sh, "r1", "nft", "delete", "table", "ip", "qos") LAB.daemon("h3", "iperf3 -s -1") wait_until(lambda: "5201" in LAB.out("h3", "netstat", "-tln"), seconds=5) traffic = self.load() self.addCleanup(traffic.wait) wait_until(lambda: "backlog 0b" not in LAB.out("r1", "tc", "-s", "qdisc", "show", "dev", "eth3"), seconds=4, step=0.2) priority = average_rtt(LAB.out("h1", "ping", "-c", "8", "-i", "0.25", "10.200.3.10")) self.assertLess(priority, loaded / 2) traffic.wait() class F_Ipv6(unittest.TestCase): @classmethod def setUpClass(cls): done = subprocess.run(["sh", os.path.join(HERE, "fwd", "ipv6.sh")], capture_output=True, text=True) if done.returncode != 0: raise RuntimeError(done.stderr) def test_hop_limit_drops_at_router(self): LAB.sh("h1", "ip", "-6", "route", "flush", "cache") before = LAB.capture("r1", "eth1", "icmp6 and ip6[40] == 128", 1, flags=("-n", "-v")) after = LAB.capture("r1", "eth3", "icmp6 and ip6[40] == 128", 1, flags=("-n", "-v")) self.assertEqual(LAB.sh("h1", "ping", "-c", "1", "-W", "2", "2001:db8:13::2").returncode, 0) self.assertIn("hlim 64", before.result()) self.assertIn("hlim 63", after.result()) def test_ping_and_neighbor_discovery(self): LAB.sh("h1", "ip", "-6", "neigh", "flush", "dev", "eth1") capture = LAB.capture("h1", "eth1", "icmp6", 4, flags=("-n", "-v")) wait_until(lambda: LAB.sh("h1", "ping", "-c", "1", "-W", "2", "2001:db8:13::2").returncode == 0, seconds=10) output = capture.result() self.assertIn("neighbor solicitation", output) self.assertRegex(output, r"> ff02::1:ff00:1\b") self.assertIn("echo request", output) def test_router_sends_packet_too_big_and_source_fragments(self): set_link_mtu(1280) self.addCleanup(set_link_mtu, 1500) LAB.sh("h1", "ip", "-6", "route", "flush", "cache") capture = LAB.capture("h1", "eth1", "icmp6 and ip6[40] != 135 and ip6[40] != 136", 2, flags=("-n", "-v")) LAB.sh("h1", "ping", "-c", "1", "-W", "1", "-s", "1400", "2001:db8:13::2") self.assertIn("packet too big, mtu 1280", capture.result()) # Ahora fragmenta el origen, con un encabezado de extensión. fragments = LAB.capture("h1", "eth1", "ip6[6] == 44", 2, flags=("-n", "-v")) done = LAB.sh("h1", "ping", "-c", "1", "-W", "2", "-s", "1400", "2001:db8:13::2") self.assertEqual(done.returncode, 0) self.assertIn("frag (", fragments.result()) class G_LabRoutes(unittest.TestCase): """Pasos 1, 2 y 6 del laboratorio. Se saltan hasta que completen lab/.""" @classmethod def setUpClass(cls): done = subprocess.run(["sh", answer_path("routes.sh")], capture_output=True, text=True) if done.returncode != 0: raise RuntimeError(done.stderr) wait_until(lambda: "10.200.2.0/24" in LAB.out("r1", "ip", "route")) def test_1_all_hosts_reach_each_other(self): addresses = {"h1": "10.200.1.10", "h2": "10.200.2.10", "h3": "10.200.3.10"} for source in addresses: for target, address in addresses.items(): if source != target: done = LAB.sh(source, "ping", "-c", "1", "-W", "2", address) self.assertEqual(done.returncode, 0, f"{source} -> {target}") def test_2_more_specific_route_wins_and_makes_path_asymmetric(self): LAB.sh("r1", "vtysh", "-c", "conf t", "-c", "ip route 10.200.3.10/32 10.200.12.2") self.addCleanup(LAB.sh, "r1", "vtysh", "-c", "conf t", "-c", "no ip route 10.200.3.10/32 10.200.12.2") self.assertTrue(wait_until(lambda: "10.200.3.10 nhid" in LAB.out("r1", "ip", "route") or "10.200.3.10 via" in LAB.out("r1", "ip", "route"))) self.assertIn("via 10.200.12.2", LAB.out("r1", "ip", "route", "get", "10.200.3.10")) self.assertIn("via 10.200.13.2", LAB.out("r1", "ip", "route", "get", "10.200.3.11")) hops = LAB.out("h1", "traceroute", "-n", "-w", "1", "10.200.3.10").splitlines()[1:] self.assertEqual(len(hops), 4) self.assertIn("10.200.12.2", hops[1]) # r3 responde con la dirección de la interfaz por la que sale hacia h1. self.assertIn("10.200.13.2", hops[2]) # A la vuelta, r3 no pasa por r2. at_r2 = LAB.capture("r2", "eth3", "icmp", 1, seconds=3) at_r3 = LAB.capture("r3", "eth3", "icmp and src 10.200.3.10", 1, seconds=3) LAB.sh("h1", "ping", "-c", "1", "-W", "2", "10.200.3.10") self.assertIn("echo request", at_r2.result()) self.assertIn("echo reply", at_r3.result()) def test_3_router_that_stays_silent_shows_asterisks(self): LAB.sh("r1", "vtysh", "-c", "conf t", "-c", "ip route 10.200.3.10/32 10.200.12.2") self.addCleanup(LAB.sh, "r1", "vtysh", "-c", "conf t", "-c", "no ip route 10.200.3.10/32 10.200.12.2") wait_until(lambda: "10.200.3.10 " in LAB.out("r1", "ip", "route")) for rule in ("add table ip quiet", "add chain ip quiet out { type filter hook output priority 0; }", "add rule ip quiet out icmp type time-exceeded drop"): self.assertEqual(LAB.sh("r3", "nft", rule).returncode, 0, rule) self.addCleanup(LAB.sh, "r3", "nft", "delete", "table", "ip", "quiet") hops = LAB.out("h1", "traceroute", "-n", "-w", "1", "10.200.3.10").splitlines()[1:] self.assertIn("* * *", hops[2].replace(" ", " ")) self.assertIn("10.200.3.10", hops[3]) def test_4_ipv6_reaches_h3(self): path = answer_path("ipv6_h3.sh") done = subprocess.run(["sh", path], capture_output=True, text=True) self.assertEqual(done.returncode, 0, done.stderr) self.assertTrue(wait_until(lambda: LAB.sh("h1", "ping", "-c", "1", "-W", "2", "2001:db8:3::10").returncode == 0, seconds=10)) match = re.search(r"ttl=(\d+)", LAB.out("h1", "ping", "-c", "1", "2001:db8:3::10")) self.assertEqual(match.group(1), "62") if __name__ == "__main__": unittest.main()