Ole Troan | 6ee3aa4 | 2023-08-17 13:36:08 +0200 | [diff] [blame^] | 1 | #!/usr/bin/env python3 |
| 2 | |
| 3 | import unittest |
| 4 | import ipaddress |
| 5 | from framework import VppTestCase, VppTestRunner |
| 6 | |
| 7 | from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest |
| 8 | from scapy.layers.l2 import Ether |
| 9 | from scapy.packet import Raw |
| 10 | |
| 11 | |
| 12 | class TestNPT66(VppTestCase): |
| 13 | """NPTv6 Test Case""" |
| 14 | |
| 15 | def setUp(self): |
| 16 | super(TestNPT66, self).setUp() |
| 17 | |
| 18 | # create 2 pg interfaces |
| 19 | self.create_pg_interfaces(range(2)) |
| 20 | |
| 21 | for i in self.pg_interfaces: |
| 22 | i.admin_up() |
| 23 | i.config_ip6() |
| 24 | i.resolve_ndp() |
| 25 | |
| 26 | def tearDown(self): |
| 27 | for i in self.pg_interfaces: |
| 28 | i.unconfig_ip6() |
| 29 | i.admin_down() |
| 30 | super(TestNPT66, self).tearDown() |
| 31 | |
| 32 | def send_and_verify(self, in2out, internal, external): |
| 33 | if in2out: |
| 34 | sendif = self.pg0 |
| 35 | recvif = self.pg1 |
| 36 | local_mac = self.pg0.local_mac |
| 37 | remote_mac = self.pg0.remote_mac |
| 38 | src = ipaddress.ip_interface(internal).ip + 1 |
| 39 | dst = self.pg1.remote_ip6 |
| 40 | else: |
| 41 | sendif = self.pg1 |
| 42 | recvif = self.pg0 |
| 43 | local_mac = self.pg1.local_mac |
| 44 | remote_mac = self.pg1.remote_mac |
| 45 | src = self.pg1.remote_ip6 |
| 46 | dst = ipaddress.ip_interface(external).ip + 1 |
| 47 | |
| 48 | p = ( |
| 49 | Ether(dst=local_mac, src=remote_mac) |
| 50 | / IPv6(src=src, dst=dst) |
| 51 | / ICMPv6EchoRequest() |
| 52 | ) |
| 53 | rxs = self.send_and_expect(sendif, p, recvif) |
| 54 | for rx in rxs: |
| 55 | rx.show2() |
| 56 | original_cksum = rx[ICMPv6EchoRequest].cksum |
| 57 | del rx[ICMPv6EchoRequest].cksum |
| 58 | rx = rx.__class__(bytes(rx)) |
| 59 | self.assertEqual(original_cksum, rx[ICMPv6EchoRequest].cksum) |
| 60 | |
| 61 | def do_test(self, internal, external): |
| 62 | self.vapi.npt66_binding_add_del( |
| 63 | sw_if_index=self.pg1.sw_if_index, |
| 64 | internal=internal, |
| 65 | external=external, |
| 66 | is_add=True, |
| 67 | ) |
| 68 | self.vapi.cli(f"ip route add {internal} via {self.pg0.remote_ip6}") |
| 69 | |
| 70 | self.send_and_verify(True, internal, external) |
| 71 | self.send_and_verify(False, internal, external) |
| 72 | |
| 73 | self.vapi.npt66_binding_add_del( |
| 74 | sw_if_index=self.pg1.sw_if_index, |
| 75 | internal=internal, |
| 76 | external=external, |
| 77 | is_add=False, |
| 78 | ) |
| 79 | |
| 80 | def test_npt66_simple(self): |
| 81 | """Send and receive a packet through NPT66""" |
| 82 | |
| 83 | self.do_test("fc00:1::/48", "2001:db8:1::/48") |
| 84 | self.do_test("fc00:1234::/32", "2001:db8:1::/32") |
| 85 | self.do_test("fc00:1234::/63", "2001:db8:1::/56") |
| 86 | |
| 87 | |
| 88 | if __name__ == "__main__": |
| 89 | unittest.main(testRunner=VppTestRunner) |