self.logger.info(self.vapi.cli("show nat64 session table all"))
-class TestDSlite(MethodHolder):
- """ DS-Lite Test Cases """
-
- @classmethod
- def setUpClass(cls):
- super(TestDSlite, cls).setUpClass()
-
- try:
- cls.nat_addr = '10.0.0.3'
-
- cls.create_pg_interfaces(range(3))
- 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(2)
- cls.pg1.configure_ipv6_neighbors()
- cls.pg2.admin_up()
- cls.pg2.config_ip4()
- cls.pg2.resolve_arp()
-
- except Exception:
- super(TestDSlite, cls).tearDownClass()
- raise
-
- @classmethod
- def tearDownClass(cls):
- super(TestDSlite, cls).tearDownClass()
-
- def verify_syslog_apmadd(self, data, isaddr, isport, xsaddr, xsport,
- sv6enc, proto):
- message = data.decode('utf-8')
- try:
- message = SyslogMessage.parse(message)
- except ParseError as e:
- self.logger.error(e)
- else:
- self.assertEqual(message.severity, SyslogSeverity.info)
- self.assertEqual(message.appname, 'NAT')
- self.assertEqual(message.msgid, 'APMADD')
- sd_params = message.sd.get('napmap')
- self.assertTrue(sd_params is not None)
- self.assertEqual(sd_params.get('IATYP'), 'IPv4')
- self.assertEqual(sd_params.get('ISADDR'), isaddr)
- self.assertEqual(sd_params.get('ISPORT'), "%d" % isport)
- self.assertEqual(sd_params.get('XATYP'), 'IPv4')
- self.assertEqual(sd_params.get('XSADDR'), xsaddr)
- self.assertEqual(sd_params.get('XSPORT'), "%d" % xsport)
- self.assertEqual(sd_params.get('PROTO'), "%d" % proto)
- self.assertTrue(sd_params.get('SSUBIX') is not None)
- self.assertEqual(sd_params.get('SV6ENC'), sv6enc)
-
- def test_dslite(self):
- """ Test DS-Lite """
- nat_config = self.vapi.nat_show_config()
- self.assertEqual(0, nat_config.dslite_ce)
-
- self.vapi.dslite_add_del_pool_addr_range(start_addr=self.nat_addr,
- end_addr=self.nat_addr,
- is_add=1)
- aftr_ip4 = '192.0.0.1'
- aftr_ip6 = '2001:db8:85a3::8a2e:370:1'
- self.vapi.dslite_set_aftr_addr(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
- self.vapi.syslog_set_sender(self.pg2.local_ip4, self.pg2.remote_ip4)
-
- # UDP
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[0].ip6) /
- IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
- 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.nat_addr)
- self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
- self.assertNotEqual(capture[UDP].sport, 20000)
- self.assertEqual(capture[UDP].dport, 10000)
- self.assert_packet_checksums_valid(capture)
- out_port = capture[UDP].sport
- capture = self.pg2.get_capture(1)
- self.verify_syslog_apmadd(capture[0][Raw].load, '192.168.1.1',
- 20000, self.nat_addr, out_port,
- self.pg1.remote_hosts[0].ip6, IP_PROTOS.udp)
-
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
- UDP(sport=10000, dport=out_port))
- 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, aftr_ip6)
- self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
- self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
- self.assertEqual(capture[IP].dst, '192.168.1.1')
- self.assertEqual(capture[UDP].sport, 10000)
- self.assertEqual(capture[UDP].dport, 20000)
- self.assert_packet_checksums_valid(capture)
-
- # TCP
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
- IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
- TCP(sport=20001, dport=10001))
- 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.nat_addr)
- self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
- self.assertNotEqual(capture[TCP].sport, 20001)
- self.assertEqual(capture[TCP].dport, 10001)
- self.assert_packet_checksums_valid(capture)
- out_port = capture[TCP].sport
-
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
- TCP(sport=10001, dport=out_port))
- 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, aftr_ip6)
- self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
- self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
- self.assertEqual(capture[IP].dst, '192.168.1.1')
- self.assertEqual(capture[TCP].sport, 10001)
- self.assertEqual(capture[TCP].dport, 20001)
- self.assert_packet_checksums_valid(capture)
-
- # ICMP
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
- IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
- ICMP(id=4000, type='echo-request'))
- 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.nat_addr)
- self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
- self.assertNotEqual(capture[ICMP].id, 4000)
- self.assert_packet_checksums_valid(capture)
- out_id = capture[ICMP].id
-
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
- ICMP(id=out_id, type='echo-reply'))
- 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, aftr_ip6)
- self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
- self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
- self.assertEqual(capture[IP].dst, '192.168.1.1')
- self.assertEqual(capture[ICMP].id, 4000)
- self.assert_packet_checksums_valid(capture)
-
- # ping DS-Lite AFTR tunnel endpoint address
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(src=self.pg1.remote_hosts[1].ip6, dst=aftr_ip6) /
- ICMPv6EchoRequest())
- self.pg1.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, aftr_ip6)
- self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
- self.assertTrue(capture.haslayer(ICMPv6EchoReply))
-
- b4s = self.statistics.get_counter('/dslite/total-b4s')
- self.assertEqual(b4s[0][0], 2)
- sessions = self.statistics.get_counter('/dslite/total-sessions')
- self.assertEqual(sessions[0][0], 3)
-
- def tearDown(self):
- super(TestDSlite, self).tearDown()
-
- def show_commands_at_teardown(self):
- self.logger.info(self.vapi.cli("show dslite pool"))
- self.logger.info(
- self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
- self.logger.info(self.vapi.cli("show dslite sessions"))
-
-
-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
-
- @classmethod
- def tearDownClass(cls):
- super(TestDSliteCE, cls).tearDownClass()
-
- def test_dslite_ce(self):
- """ Test DS-Lite CE """
-
- nat_config = self.vapi.nat_show_config()
- self.assertEqual(1, nat_config.dslite_ce)
-
- b4_ip4 = '192.0.0.2'
- b4_ip6 = '2001:db8:62aa::375e:f4c1:1'
- self.vapi.dslite_set_b4_addr(ip4_addr=b4_ip4, ip6_addr=b4_ip6)
-
- aftr_ip4 = '192.0.0.1'
- 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(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
-
- r1 = VppIpRoute(self, aftr_ip6, 128,
- [VppRoutePath(self.pg1.remote_ip6,
- self.pg1.sw_if_index)])
- r1.add_vpp_config()
-
- # 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.assert_packet_checksums_valid(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.assert_packet_checksums_valid(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)
- 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()
-
- def show_commands_at_teardown(self):
- 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 """
Code Review / vpp.git / blobdiff