"""Pruebas de hijacks, fugas y defensas en bgp/bgp.clab.yml.""" import json import os import re import subprocess import unittest import scenarios from netlab import ANSWERS, HERE, answer_path, wait_until LAB = scenarios.LAB LEGIT = "servidor legitimo" ATTACKER = "ATACANTE" SERVER = scenarios.SERVER VICTIM = scenarios.VICTIM HALF = scenarios.HALVES[0] def setUpModule(): unittest.addModuleCleanup(LAB.destroy) LAB.deploy() if not wait_until(baseline, seconds=60): raise RuntimeError("la topología no quedó lista") def web(node, address=SERVER): return LAB.out(node, "curl", "-s", "-m", "3", f"http://{address}:8080/") def route(node, prefix): """(AS_PATH de la mejor ruta, estado RPKI), o None si no está en la tabla.""" text = LAB.out(node, "vtysh", "-c", f"show bgp ipv4 unicast {prefix} json") try: for path in json.loads(text).get("paths", []): if path.get("bestpath", {}).get("overall"): return path["aspath"]["string"], path.get("rpkiValidationState", "") except ValueError: pass return None def path_of(node, prefix): found = route(node, prefix) return found[0] if found else None def origin(node, prefix): path = path_of(node, prefix) return path.split()[-1] if path else None def trace(node, address): return re.findall(r"^\s*\d+\s+(\S+)", LAB.out(node, "traceroute", "-n", "-w", "1", address), re.MULTILINE) def baseline(): return ("64630 64640 64620 64603" == path_of("a", VICTIM) and "64640 64630 64610 64601" == path_of("b", "100.64.1.0/24") and route("a", HALF) is None and route("d", HALF) is None and route("a", "100.64.4.0/24") is not None and route("b", "100.64.4.0/24") is not None and LEGIT in web("h1")) class Scenario(unittest.TestCase): def setUp(self): self.assertTrue(wait_until(baseline, seconds=45), "no se volvió al estado inicial") self.addCleanup(scenarios.reset) def wait(self, predicate, seconds=25): self.assertTrue(wait_until(predicate, seconds=seconds), "no ocurrió a tiempo") class A_Baseline(Scenario): def test_host_reaches_the_server_along_the_hierarchy(self): self.assertIn(LEGIT, web("h1")) self.assertEqual(trace("h1", SERVER), ["100.64.1.1", "10.207.4.2", "10.207.2.2", "10.207.1.2", "10.207.3.1", "10.207.8.1", SERVER]) def test_multihomed_customer_exports_only_its_own_prefix(self): self.assertNotIn("64604", path_of("b", "100.64.1.0/24")) self.assertNotIn("64604", path_of("a", VICTIM)) class B_SamePrefix(Scenario): def test_some_ases_are_fooled_and_others_are_not(self): scenarios.same() self.wait(lambda: origin("r1", VICTIM) == "64666") self.assertEqual(path_of("a", VICTIM), "64666") self.assertEqual(path_of("c", VICTIM), "64610 64666") # Del otro lado de la cima, el camino legítimo sigue siendo más corto. self.assertEqual(path_of("d", VICTIM), "64620 64603") self.assertEqual(path_of("b", VICTIM), "64603") self.assertIn(ATTACKER, web("h1")) self.assertEqual(trace("h1", SERVER)[-1], SERVER) self.assertEqual(len(trace("h1", SERVER)), 3) def test_withdrawing_heals_the_routing(self): scenarios.same() self.wait(lambda: ATTACKER in web("h1")) scenarios.reset() self.wait(lambda: LEGIT in web("h1")) class C_MoreSpecific(Scenario): def test_everyone_goes_to_the_attacker_for_the_covered_half(self): scenarios.specific() self.wait(lambda: route("r3", HALF) is not None) for node in ("r1", "a", "c", "d", "b", "r4", "r3"): with self.subTest(node=node): self.assertEqual(origin(node, HALF), "64666") self.assertIn("100.64.3.0/25", LAB.out(node, "vtysh", "-c", f"show ip route {SERVER}")) self.assertIn(ATTACKER, web("h1")) def test_the_other_half_still_reaches_the_victim(self): scenarios.specific() self.wait(lambda: ATTACKER in web("h1")) self.assertIn(LEGIT, web("h1", "100.64.3.200")) def test_victim_can_win_back_with_longer_prefixes(self): scenarios.same() self.wait(lambda: ATTACKER in web("h1")) scenarios.counter() self.wait(lambda: LEGIT in web("h1")) self.assertEqual(origin("r1", HALF), "64603") self.assertEqual(origin("r1", VICTIM), "64666") class D_Leak(Scenario): def test_leak_makes_the_customer_carry_traffic_between_its_providers(self): scenarios.leak() self.wait(lambda: path_of("b", "100.64.1.0/24") == "64604 64610 64601") self.assertEqual(path_of("a", VICTIM), "64604 64620 64603") self.assertIn("10.207.7.1", trace("srv", "100.64.1.10")) self.assertIn("10.207.6.1", trace("h1", SERVER)) def test_leak_adds_the_delay_of_the_small_network(self): scenarios.slow_r4() scenarios.leak() self.wait(lambda: path_of("b", "100.64.1.0/24") == "64604 64610 64601") output = LAB.out("srv", "ping", "-c", "5", "-i", "0.2", "100.64.1.10") average = float(re.search(r"= [\d.]+/([\d.]+)/", output).group(1)) self.assertGreater(average, 8) def test_max_prefix_closes_the_session_with_the_leaker(self): scenarios.max_prefix() scenarios.leak() self.wait(lambda: "Idle (PfxCt)" in LAB.out("b", "vtysh", "-c", "show bgp summary")) self.wait(lambda: path_of("b", "100.64.1.0/24") == "64640 64630 64610 64601") # El precio: b ya no aprende la red de r4 por esa sesión, hasta que alguien la reabra. self.assertEqual(path_of("b", "100.64.4.0/24"), "64640 64630 64610 64604") class E_Rov(Scenario): def rov_state(self, node, prefix): found = route(node, prefix) return found[1] if found else None def wait_table(self, node): self.wait(lambda: "Number of IPv4 Prefixes: 2" in LAB.out(node, "vtysh", "-c", "show rpki prefix-table")) def test_invalid_is_dropped_only_when_policy_says_so(self): # Sin la regla, el router marca la ruta pero la sigue usando (RFC 6811, sec. 2). scenarios.conf("a", "rpki", "rpki polling_period 5", "rpki cache tcp 10.207.9.2 8282 preference 1") self.wait_table("a") scenarios.same() self.wait(lambda: self.rov_state("a", VICTIM) == "invalid") self.assertEqual(path_of("a", VICTIM), "64666") self.assertIn(ATTACKER, web("h1")) scenarios.rov_a() self.wait(lambda: self.rov_state("a", VICTIM) == "valid") self.assertEqual(origin("a", VICTIM), "64603") self.wait(lambda: LEGIT in web("h1")) def test_rov_matches_the_classifier(self): scenarios.conf("a", "rpki", "rpki polling_period 5", "rpki cache tcp 10.207.9.2 8282 preference 1") self.wait_table("a") scenarios.specific() self.wait(lambda: route("a", HALF) is not None) vrps = os.path.join(HERE, "bgp", "cache", "vrps.txt") for prefix, asn, state in ((HALF, 64666, "invalid"), ("100.64.1.0/24", 64601, "valid"), ("100.64.66.0/24", 64666, "not found")): with self.subTest(prefix=prefix): self.wait(lambda: self.rov_state("a", prefix) == state.replace("not found", "notfound") or self.rov_state("a", prefix) == state) done = subprocess.run([os.path.join(HERE, "rov"), vrps, prefix, str(asn)], capture_output=True, text=True, check=True) self.assertEqual(done.stdout.strip().splitlines()[-1], state) def test_more_specific_is_invalid_by_max_length(self): scenarios.rov_a() self.wait_table("a") scenarios.specific() self.wait(lambda: route("r6", HALF) is not None) self.assertFalse(wait_until(lambda: route("a", HALF) is not None, seconds=4)) self.assertIn(LEGIT, web("h1")) def test_forged_origin_passes_rov(self): scenarios.rov_a() self.wait_table("a") scenarios.forged() self.wait(lambda: path_of("a", VICTIM) == "64666 64603") self.assertEqual(self.rov_state("a", VICTIM), "valid") self.assertIn(ATTACKER, web("h1")) def test_forged_more_specific_depends_on_the_max_length_of_the_roa(self): scenarios.rov_a() self.wait_table("a") scenarios.forged_specific() self.wait(lambda: route("r6", HALF) is not None) self.assertFalse(wait_until(lambda: route("a", HALF) is not None, seconds=4)) self.assertIn(LEGIT, web("h1")) def test_lax_max_length_lets_the_forged_more_specific_through(self): scenarios.rov_a_lax() self.wait_table("a") self.wait(lambda: "24 - 25" in LAB.out("a", "vtysh", "-c", "show rpki prefix-table")) scenarios.forged_specific() self.wait(lambda: path_of("a", HALF) == "64666 64603") self.assertEqual(self.rov_state("a", HALF), "valid") self.assertIn(ATTACKER, web("h1")) def test_partial_deployment_protects_only_those_behind_it(self): scenarios.rov_d() self.wait_table("d") scenarios.specific() self.wait(lambda: route("c", HALF) is not None) self.assertFalse(wait_until(lambda: route("d", HALF) is not None, seconds=4)) self.assertIsNone(route("b", HALF)) self.assertIn(ATTACKER, web("h1")) def test_rov_does_not_stop_a_leak(self): scenarios.rov_a() scenarios.rov_d() self.wait_table("a") self.wait_table("d") scenarios.leak() self.wait(lambda: path_of("a", VICTIM) == "64604 64620 64603") self.assertEqual(self.rov_state("a", VICTIM), "valid") class F_Filters(Scenario): """Paso 5 del laboratorio. Se salta hasta que completen lab/filters-*.conf.""" def setUp(self): answer_path("filters-a.conf") answer_path("filters-b.conf") super().setUp() scenarios.filters(ANSWERS) self.assertTrue(wait_until(baseline, seconds=30), "los filtros rompieron rutas legítimas") def assert_attacker_kept_out(self): self.assertFalse(wait_until(lambda: ATTACKER in web("h1"), seconds=4)) self.assertEqual(origin("a", VICTIM), "64603") self.assertIsNone(route("a", HALF)) self.assertIn(LEGIT, web("h1")) def test_legitimate_prefixes_still_pass(self): for node, prefix, asn in (("a", "100.64.1.0/24", "64601"), ("a", "100.64.66.0/24", "64666"), ("a", "100.64.4.0/24", "64604"), ("b", "100.64.3.0/24", "64603"), ("b", "100.64.4.0/24", "64604")): with self.subTest(node=node, prefix=prefix): self.assertEqual(origin(node, prefix), asn) def test_same_prefix_is_rejected(self): scenarios.same() self.assertTrue(wait_until(lambda: route("r6", VICTIM) is not None)) self.assert_attacker_kept_out() def test_more_specific_is_rejected(self): scenarios.specific() self.assertTrue(wait_until(lambda: route("r6", HALF) is not None)) self.assert_attacker_kept_out() def test_forged_origin_is_rejected(self): scenarios.forged() self.assertTrue(wait_until(lambda: route("r6", VICTIM) is not None)) self.assert_attacker_kept_out() def test_more_specific_of_the_own_block_is_rejected(self): scenarios.conf("r6", "ip route 100.64.66.0/25 Null0", *scenarios.in_bgp(64666, "network 100.64.66.0/25")) self.addCleanup(lambda: (scenarios.conf("r6", *scenarios.in_bgp(64666, "no network 100.64.66.0/25")), scenarios.conf("r6", "no ip route 100.64.66.0/25 Null0"))) self.assertTrue(wait_until(lambda: route("r6", "100.64.66.0/25") is not None)) self.assertFalse(wait_until(lambda: route("a", "100.64.66.0/25") is not None, seconds=4)) def test_leak_is_rejected_on_both_sides(self): scenarios.leak() self.assertFalse(wait_until(lambda: "64604" in (path_of("b", "100.64.1.0/24") or ""), seconds=4)) self.assertNotIn("64604", path_of("a", VICTIM)) self.assertNotIn("10.207.7.1", trace("srv", "100.64.1.10")) if __name__ == "__main__": unittest.main()