+
+class TestDSliteCE(MethodHolder):
+ """ DS-Lite CE Test Cases """
+
+ @classmethod
+ def setUpConstants(cls):
+ super(TestDSliteCE, cls).setUpConstants()
+ cls.vpp_cmdline.extend(["nat", "{", "dslite ce", "}"])
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestDSliteCE, cls).setUpClass()
+
+ try:
+ cls.create_pg_interfaces(range(2))
+ cls.pg0.admin_up()
+ cls.pg0.config_ip4()
+ cls.pg0.resolve_arp()
+ cls.pg1.admin_up()
+ cls.pg1.config_ip6()
+ cls.pg1.generate_remote_hosts(1)
+ cls.pg1.configure_ipv6_neighbors()
+
+ except Exception:
+ super(TestDSliteCE, cls).tearDownClass()
+ raise
+
+ def test_dslite_ce(self):
+ """ Test DS-Lite CE """
+
+ b4_ip4 = '192.0.0.2'
+ b4_ip4_n = socket.inet_pton(socket.AF_INET, b4_ip4)
+ b4_ip6 = '2001:db8:62aa::375e:f4c1:1'
+ b4_ip6_n = socket.inet_pton(socket.AF_INET6, b4_ip6)
+ self.vapi.dslite_set_b4_addr(b4_ip6_n, b4_ip4_n)
+
+ aftr_ip4 = '192.0.0.1'
+ aftr_ip4_n = socket.inet_pton(socket.AF_INET, aftr_ip4)
+ aftr_ip6 = '2001:db8:85a3::8a2e:370:1'
+ aftr_ip6_n = socket.inet_pton(socket.AF_INET6, aftr_ip6)
+ self.vapi.dslite_set_aftr_addr(aftr_ip6_n, aftr_ip4_n)
+
+ self.vapi.ip_add_del_route(dst_address=aftr_ip6_n,
+ dst_address_length=128,
+ next_hop_address=self.pg1.remote_ip6n,
+ next_hop_sw_if_index=self.pg1.sw_if_index,
+ is_ipv6=1)
+
+ # UDP encapsulation
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(dst=self.pg1.remote_ip4, src=self.pg0.remote_ip4) /
+ UDP(sport=10000, dport=20000))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ capture = capture[0]
+ self.assertEqual(capture[IPv6].src, b4_ip6)
+ self.assertEqual(capture[IPv6].dst, aftr_ip6)
+ self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
+ self.assertEqual(capture[IP].dst, self.pg1.remote_ip4)
+ self.assertEqual(capture[UDP].sport, 10000)
+ self.assertEqual(capture[UDP].dport, 20000)
+ self.check_ip_checksum(capture)
+
+ # UDP decapsulation
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IPv6(dst=b4_ip6, src=aftr_ip6) /
+ IP(dst=self.pg0.remote_ip4, src=self.pg1.remote_ip4) /
+ UDP(sport=20000, dport=10000))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(1)
+ capture = capture[0]
+ self.assertFalse(capture.haslayer(IPv6))
+ self.assertEqual(capture[IP].src, self.pg1.remote_ip4)
+ self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
+ self.assertEqual(capture[UDP].sport, 20000)
+ self.assertEqual(capture[UDP].dport, 10000)
+ self.check_ip_checksum(capture)
+
+ # ping DS-Lite B4 tunnel endpoint address
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IPv6(src=self.pg1.remote_hosts[0].ip6, dst=b4_ip6) /
+ ICMPv6EchoRequest())
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ self.assertEqual(1, len(capture))
+ capture = capture[0]
+ self.assertEqual(capture[IPv6].src, b4_ip6)
+ self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
+ self.assertTrue(capture.haslayer(ICMPv6EchoReply))
+
+ def tearDown(self):
+ super(TestDSliteCE, self).tearDown()
+ if not self.vpp_dead:
+ self.logger.info(
+ self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
+ self.logger.info(
+ self.vapi.cli("show dslite b4-tunnel-endpoint-address"))
+
+
+class TestNAT66(MethodHolder):
+ """ NAT66 Test Cases """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestNAT66, cls).setUpClass()
+
+ try:
+ cls.nat_addr = 'fd01:ff::2'
+ cls.nat_addr_n = socket.inet_pton(socket.AF_INET6, cls.nat_addr)
+
+ cls.create_pg_interfaces(range(2))
+ cls.interfaces = list(cls.pg_interfaces)
+
+ for i in cls.interfaces:
+ i.admin_up()
+ i.config_ip6()
+ i.configure_ipv6_neighbors()
+
+ except Exception:
+ super(TestNAT66, cls).tearDownClass()
+ raise
+
+ def test_static(self):
+ """ 1:1 NAT66 test """
+ self.vapi.nat66_add_del_interface(self.pg0.sw_if_index)
+ self.vapi.nat66_add_del_interface(self.pg1.sw_if_index, is_inside=0)
+ self.vapi.nat66_add_del_static_mapping(self.pg0.remote_ip6n,
+ self.nat_addr_n)
+
+ # in2out
+ pkts = []
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
+ TCP())
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
+ UDP())
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
+ ICMPv6EchoRequest())
+ pkts.append(p)
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
+ GRE() / IP() / TCP())
+ pkts.append(p)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IPv6].src, self.nat_addr)
+ self.assertEqual(packet[IPv6].dst, self.pg1.remote_ip6)
+ if packet.haslayer(TCP):
+ self.check_tcp_checksum(packet)
+ elif packet.haslayer(UDP):
+ self.check_udp_checksum(packet)
+ elif packet.haslayer(ICMPv6EchoRequest):
+ self.check_icmpv6_checksum(packet)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ # out2in
+ pkts = []
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
+ TCP())
+ pkts.append(p)
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
+ UDP())
+ pkts.append(p)
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
+ ICMPv6EchoReply())
+ pkts.append(p)
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
+ GRE() / IP() / TCP())
+ pkts.append(p)
+ self.pg1.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ for packet in capture:
+ try:
+ self.assertEqual(packet[IPv6].src, self.pg1.remote_ip6)
+ self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6)
+ if packet.haslayer(TCP):
+ self.check_tcp_checksum(packet)
+ elif packet.haslayer(UDP):
+ self.check_udp_checksum(packet)
+ elif packet.haslayer(ICMPv6EchoReply):
+ self.check_icmpv6_checksum(packet)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ sm = self.vapi.nat66_static_mapping_dump()
+ self.assertEqual(len(sm), 1)
+ self.assertEqual(sm[0].total_pkts, 8)
+
+ def test_check_no_translate(self):
+ """ NAT66 translate only when egress interface is outside interface """
+ self.vapi.nat66_add_del_interface(self.pg0.sw_if_index)
+ self.vapi.nat66_add_del_interface(self.pg1.sw_if_index)
+ self.vapi.nat66_add_del_static_mapping(self.pg0.remote_ip6n,
+ self.nat_addr_n)
+
+ # in2out
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
+ UDP())
+ self.pg0.add_stream([p])
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ packet = capture[0]
+ try:
+ self.assertEqual(packet[IPv6].src, self.pg0.remote_ip6)
+ self.assertEqual(packet[IPv6].dst, self.pg1.remote_ip6)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", packet))
+ raise
+
+ def clear_nat66(self):
+ """
+ Clear NAT66 configuration.
+ """
+ interfaces = self.vapi.nat66_interface_dump()
+ for intf in interfaces:
+ self.vapi.nat66_add_del_interface(intf.sw_if_index,
+ intf.is_inside,
+ is_add=0)
+
+ static_mappings = self.vapi.nat66_static_mapping_dump()
+ for sm in static_mappings:
+ self.vapi.nat66_add_del_static_mapping(sm.local_ip_address,
+ sm.external_ip_address,
+ sm.vrf_id,
+ is_add=0)
+
+ def tearDown(self):
+ super(TestNAT66, self).tearDown()
+ if not self.vpp_dead:
+ self.logger.info(self.vapi.cli("show nat66 interfaces"))
+ self.logger.info(self.vapi.cli("show nat66 static mappings"))
+ self.clear_nat66()
+