"""Pruebas de los seis AS sin políticas (ases/ases.clab.yml): sesiones, mensajes y el problema.""" import os import shutil import subprocess import tempfile import time import unittest import spvp from netlab import HERE, Lab, run, wait_until LAB = Lab("ases", "w06-ases", "redes-w06-ases") ROUTERS = ["r1", "r2", "r3", "r4", "r5", "r6"] PREFIXES = [f"100.64.{n}.0/24" for n in range(1, 7)] SESSIONS = {"r1": 2, "r2": 2, "r3": 1, "r4": 2, "r5": 2, "r6": 3} def setUpModule(): unittest.addModuleCleanup(LAB.destroy) LAB.deploy() if not LAB.wait_for_table(ROUTERS, PREFIXES): raise RuntimeError("la topología no quedó lista") class A_Sessions(unittest.TestCase): def test_every_session_is_established(self): for node, count in SESSIONS.items(): self.assertTrue(wait_until(lambda n=node, c=count: LAB.sessions_up(n, c), seconds=30), node) def test_as_path_lists_who_the_route_crossed(self): self.assertEqual(LAB.best_path("r3", "100.64.3.0/24"), []) self.assertEqual(LAB.best_path("r1", "100.64.3.0/24"), [64604, 64606, 64603]) self.assertEqual(LAB.best_path("r5", "100.64.3.0/24"), [64606, 64603]) def test_shortest_as_path_decides_and_the_router_says_so(self): text = LAB.vtysh("r1", "show bgp ipv4 100.64.3.0/24") self.assertIn("64602 64605 64606 64603", text) self.assertIn("best (AS Path)", text) def test_hosts_reach_each_other(self): self.assertTrue(wait_until(lambda: LAB.ok("h1", "ping", "-c", "1", "-W", "1", "100.64.3.10"))) def test_unknown_destination_fails(self): # Sin ruta por defecto, lo desconocido falla en vez de escaparse por Docker. self.assertFalse(LAB.ok("r1", "ping", "-c", "1", "-W", "1", "198.51.100.1")) class B_Messages(unittest.TestCase): """Se apaga una sesión, se captura mientras vuelve y se leen los mensajes.""" @classmethod def setUpClass(cls): peer = "100.64.2.129" LAB.vtysh("r1", "configure terminal", "router bgp 64601", f"neighbor {peer} shutdown") time.sleep(1) LAB.daemon("r1", "timeout 14 tcpdump -U -i eth2 -w /tmp/session.pcap tcp port 179") time.sleep(1.5) LAB.vtysh("r1", "configure terminal", "router bgp 64601", f"no neighbor {peer} shutdown") wait_until(lambda: LAB.sessions_up("r1", 2), seconds=20) time.sleep(4) LAB.vtysh("r1", "configure terminal", "router bgp 64601", f"neighbor {peer} shutdown") time.sleep(1.5) LAB.vtysh("r1", "configure terminal", "router bgp 64601", f"no neighbor {peer} shutdown") wait_until(lambda: LAB.sessions_up("r1", 2), seconds=20) time.sleep(1) LAB.out("r1", "pkill", "tcpdump") time.sleep(1) cls.folder = tempfile.mkdtemp(dir=HERE, prefix=".capture-") cls.pcap = os.path.join(cls.folder, "session.pcap") done = run(["docker", "cp", f"{LAB.prefix}r1:/tmp/session.pcap", cls.pcap]) if done.returncode != 0: raise RuntimeError(done.stderr) @classmethod def tearDownClass(cls): shutil.rmtree(cls.folder, ignore_errors=True) def fields(self, display_filter, *names): command = ["tshark", "-r", self.pcap, "-Y", display_filter, "-T", "fields"] for name in names: command += ["-e", name] done = subprocess.run(command, capture_output=True, text=True, timeout=60) return [line.split("\t") for line in done.stdout.splitlines() if line.strip()] def test_tcp_port_179(self): self.assertTrue(self.fields("tcp.flags.syn == 1 and tcp.flags.ack == 0 and tcp.dstport == 179", "ip.src")) def test_open_from_both_sides_with_as_and_hold_time(self): rows = self.fields("bgp.type == 1", "ip.src", "bgp.open.myas", "bgp.open.holdtime", "bgp.open.version", "bgp.open.identifier") by_source = {row[0]: row for row in rows} self.assertEqual(by_source["100.64.2.129"][1:], ["64602", "180", "4", "100.64.2.1"]) self.assertEqual(by_source["100.64.2.130"][1:], ["64601", "180", "4", "100.64.1.1"]) def test_bgp_messages_travel_with_ttl_1_on_ebgp(self): rows = self.fields("bgp", "ip.ttl") self.assertTrue(rows) self.assertEqual({row[0] for row in rows}, {"1"}) def test_keepalive_is_only_a_header(self): rows = self.fields("bgp.type == 4", "bgp.length") self.assertTrue(rows) self.assertEqual({row[0] for row in rows}, {"19"}) def test_update_carries_path_and_next_hop(self): done = subprocess.run(["tshark", "-r", self.pcap, "-Y", "bgp.type == 2 and ip.src == 100.64.2.130", "-O", "bgp"], capture_output=True, text=True, timeout=60) self.assertIn("AS_PATH: 64601", done.stdout) self.assertIn("NEXT_HOP: 100.64.2.130", done.stdout) self.assertIn("100.64.1.0/24", done.stdout) def test_loop_prevention_a_route_with_my_own_as_is_ignored(self): output = LAB.out("r1", "tcpdump", "-nr", "/tmp/session.pcap", "-vv", "src host 100.64.2.129") self.assertRegex(output, r"AS Path[^\n]*64601") for path in LAB.bgp("r1").get("100.64.1.0/24", []): self.assertNotIn("64601", path.get("path", "")) def test_shutdown_sends_notification_cease(self): rows = self.fields("bgp.type == 3", "bgp.notify.major_error") self.assertTrue(rows) self.assertEqual(rows[0][0], "6") class C_FreeTransit(unittest.TestCase): """El paso 3 del laboratorio: r1, el AS más chico, le da tránsito a sus dos proveedores.""" @classmethod def setUpClass(cls): LAB.wait_for_table(ROUTERS, PREFIXES) wait_until(lambda: LAB.best_path("r2", "100.64.4.0/24") == [64601, 64604], seconds=30) def test_r2_reaches_its_neighbour_provider_through_its_customer(self): self.assertEqual(LAB.best_path("r2", "100.64.4.0/24"), [64601, 64604]) self.assertEqual(LAB.best_path("r4", "100.64.2.0/24"), [64601, 64602]) def test_the_path_has_a_valley(self): graph = spvp.ases_graph() self.assertFalse(spvp.is_valley_free(graph, [64602, 64601, 64604])) def test_r1_announces_each_provider_to_the_other(self): self.assertIn("100.64.4.0/24", LAB.advertised("r1", "100.64.2.129")) self.assertIn("100.64.2.0/24", LAB.advertised("r1", "100.64.4.129")) def test_traffic_really_crosses_r1(self): LAB.assert_hops(self, "r2", "100.64.4.1", ["100.64.2.130", "100.64.4.1"], source="100.64.2.1") def test_a_host_uses_the_shorter_of_its_two_providers(self): LAB.assert_hops(self, "h1", "100.64.3.10", ["100.64.1.1", "100.64.4.129", "100.64.6.129", "100.64.6.134", "100.64.3.10"]) class D_Failure(unittest.TestCase): def test_withdraw_moves_traffic_and_recovery_brings_it_back(self): self.addCleanup(LAB.out, "r1", "ip", "link", "set", "eth3", "up") LAB.out("r1", "ip", "link", "set", "eth3", "down") moved = lambda: LAB.best_path("r1", "100.64.3.0/24") == [64602, 64605, 64606, 64603] self.assertTrue(wait_until(moved, seconds=20)) self.assertTrue(wait_until(lambda: LAB.ok("h1", "ping", "-c", "1", "-W", "1", "100.64.3.10"), seconds=15)) LAB.out("r1", "ip", "link", "set", "eth3", "up") back = lambda: LAB.best_path("r1", "100.64.3.0/24") == [64604, 64606, 64603] self.assertTrue(wait_until(back, seconds=60)) class E_Anycast(unittest.TestCase): """Dos AS originan el mismo prefijo. Cada quien llega a la instancia más cercana.""" SCRIPT = os.path.join(HERE, "ases", "anycast.sh") def answer(self, node): return LAB.out(node, "sh", "-c", "nc -w 2 100.64.200.1 5300