8 from framework import tag_fixme_vpp_workers
9 from framework import VppTestCase, VppTestRunner
12 from scapy.layers.inet import IP, TCP, UDP, ICMP
13 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
14 from scapy.layers.inet6 import (
23 from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6, IPv6ExtHdrFragment
24 from scapy.layers.l2 import Ether, ARP, GRE
25 from scapy.data import IP_PROTOS
26 from scapy.packet import bind_layers, Raw
28 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
29 from time import sleep
30 from util import ip4_range
31 from vpp_papi import mac_pton
32 from syslog_rfc5424_parser import SyslogMessage, ParseError
33 from syslog_rfc5424_parser.constants import SyslogFacility, SyslogSeverity
34 from io import BytesIO
35 from vpp_papi import VppEnum
36 from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathType
37 from vpp_neighbor import VppNeighbor
38 from scapy.all import (
51 from ipaddress import IPv6Network
54 @tag_fixme_vpp_workers
55 class TestDSlite(VppTestCase):
56 """DS-Lite Test Cases"""
60 super(TestDSlite, cls).setUpClass()
63 cls.nat_addr = "10.0.0.3"
65 cls.create_pg_interfaces(range(3))
71 cls.pg1.generate_remote_hosts(2)
72 cls.pg1.configure_ipv6_neighbors()
78 super(TestDSlite, cls).tearDownClass()
82 def tearDownClass(cls):
83 super(TestDSlite, cls).tearDownClass()
85 def verify_syslog_apmadd(self, data, isaddr, isport, xsaddr, xsport, sv6enc, proto):
86 message = data.decode("utf-8")
88 message = SyslogMessage.parse(message)
89 except ParseError as e:
92 self.assertEqual(message.severity, SyslogSeverity.info)
93 self.assertEqual(message.appname, "NAT")
94 self.assertEqual(message.msgid, "APMADD")
95 sd_params = message.sd.get("napmap")
96 self.assertTrue(sd_params is not None)
97 self.assertEqual(sd_params.get("IATYP"), "IPv4")
98 self.assertEqual(sd_params.get("ISADDR"), isaddr)
99 self.assertEqual(sd_params.get("ISPORT"), "%d" % isport)
100 self.assertEqual(sd_params.get("XATYP"), "IPv4")
101 self.assertEqual(sd_params.get("XSADDR"), xsaddr)
102 self.assertEqual(sd_params.get("XSPORT"), "%d" % xsport)
103 self.assertEqual(sd_params.get("PROTO"), "%d" % proto)
104 self.assertTrue(sd_params.get("SSUBIX") is not None)
105 self.assertEqual(sd_params.get("SV6ENC"), sv6enc)
107 def test_dslite(self):
109 self.vapi.dslite_add_del_pool_addr_range(
110 start_addr=self.nat_addr, end_addr=self.nat_addr, is_add=1
112 aftr_ip4 = "192.0.0.1"
113 aftr_ip6 = "2001:db8:85a3::8a2e:370:1"
114 self.vapi.dslite_set_aftr_addr(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
115 self.vapi.syslog_set_sender(self.pg2.local_ip4, self.pg2.remote_ip4)
119 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
120 / IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[0].ip6)
121 / IP(dst=self.pg0.remote_ip4, src="192.168.1.1")
122 / UDP(sport=20000, dport=10000)
124 self.pg1.add_stream(p)
125 self.pg_enable_capture(self.pg_interfaces)
127 capture = self.pg0.get_capture(1)
129 self.assertFalse(capture.haslayer(IPv6))
130 self.assertEqual(capture[IP].src, self.nat_addr)
131 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
132 self.assertNotEqual(capture[UDP].sport, 20000)
133 self.assertEqual(capture[UDP].dport, 10000)
134 self.assert_packet_checksums_valid(capture)
135 out_port = capture[UDP].sport
136 capture = self.pg2.get_capture(1)
137 self.verify_syslog_apmadd(
138 capture[0][Raw].load,
143 self.pg1.remote_hosts[0].ip6,
148 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
149 / IP(dst=self.nat_addr, src=self.pg0.remote_ip4)
150 / UDP(sport=10000, dport=out_port)
152 self.pg0.add_stream(p)
153 self.pg_enable_capture(self.pg_interfaces)
155 capture = self.pg1.get_capture(1)
157 self.assertEqual(capture[IPv6].src, aftr_ip6)
158 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
159 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
160 self.assertEqual(capture[IP].dst, "192.168.1.1")
161 self.assertEqual(capture[UDP].sport, 10000)
162 self.assertEqual(capture[UDP].dport, 20000)
163 self.assert_packet_checksums_valid(capture)
167 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
168 / IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6)
169 / IP(dst=self.pg0.remote_ip4, src="192.168.1.1")
170 / TCP(sport=20001, dport=10001)
172 self.pg1.add_stream(p)
173 self.pg_enable_capture(self.pg_interfaces)
175 capture = self.pg0.get_capture(1)
177 self.assertFalse(capture.haslayer(IPv6))
178 self.assertEqual(capture[IP].src, self.nat_addr)
179 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
180 self.assertNotEqual(capture[TCP].sport, 20001)
181 self.assertEqual(capture[TCP].dport, 10001)
182 self.assert_packet_checksums_valid(capture)
183 out_port = capture[TCP].sport
186 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
187 / IP(dst=self.nat_addr, src=self.pg0.remote_ip4)
188 / TCP(sport=10001, dport=out_port)
190 self.pg0.add_stream(p)
191 self.pg_enable_capture(self.pg_interfaces)
193 capture = self.pg1.get_capture(1)
195 self.assertEqual(capture[IPv6].src, aftr_ip6)
196 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
197 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
198 self.assertEqual(capture[IP].dst, "192.168.1.1")
199 self.assertEqual(capture[TCP].sport, 10001)
200 self.assertEqual(capture[TCP].dport, 20001)
201 self.assert_packet_checksums_valid(capture)
205 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
206 / IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6)
207 / IP(dst=self.pg0.remote_ip4, src="192.168.1.1")
208 / ICMP(id=4000, type="echo-request")
210 self.pg1.add_stream(p)
211 self.pg_enable_capture(self.pg_interfaces)
213 capture = self.pg0.get_capture(1)
215 self.assertFalse(capture.haslayer(IPv6))
216 self.assertEqual(capture[IP].src, self.nat_addr)
217 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
218 self.assertNotEqual(capture[ICMP].id, 4000)
219 self.assert_packet_checksums_valid(capture)
220 out_id = capture[ICMP].id
223 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
224 / IP(dst=self.nat_addr, src=self.pg0.remote_ip4)
225 / ICMP(id=out_id, type="echo-reply")
227 self.pg0.add_stream(p)
228 self.pg_enable_capture(self.pg_interfaces)
230 capture = self.pg1.get_capture(1)
232 self.assertEqual(capture[IPv6].src, aftr_ip6)
233 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
234 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
235 self.assertEqual(capture[IP].dst, "192.168.1.1")
236 self.assertEqual(capture[ICMP].id, 4000)
237 self.assert_packet_checksums_valid(capture)
239 # ping DS-Lite AFTR tunnel endpoint address
241 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
242 / IPv6(src=self.pg1.remote_hosts[1].ip6, dst=aftr_ip6)
243 / ICMPv6EchoRequest()
245 self.pg1.add_stream(p)
246 self.pg_enable_capture(self.pg_interfaces)
248 capture = self.pg1.get_capture(1)
250 self.assertEqual(capture[IPv6].src, aftr_ip6)
251 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
252 self.assertTrue(capture.haslayer(ICMPv6EchoReply))
254 b4s = self.statistics.get_counter("/dslite/total-b4s")
255 self.assertEqual(b4s[0][0], 2)
256 sessions = self.statistics.get_counter("/dslite/total-sessions")
257 self.assertEqual(sessions[0][0], 3)
260 super(TestDSlite, self).tearDown()
262 def show_commands_at_teardown(self):
263 self.logger.info(self.vapi.cli("show dslite pool"))
264 self.logger.info(self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
265 self.logger.info(self.vapi.cli("show dslite sessions"))
268 class TestDSliteCE(VppTestCase):
269 """DS-Lite CE Test Cases"""
272 def setUpConstants(cls):
273 super(TestDSliteCE, cls).setUpConstants()
274 cls.vpp_cmdline.extend(["dslite", "{", "ce", "}"])
278 super(TestDSliteCE, cls).setUpClass()
281 cls.create_pg_interfaces(range(2))
284 cls.pg0.resolve_arp()
287 cls.pg1.generate_remote_hosts(1)
288 cls.pg1.configure_ipv6_neighbors()
291 super(TestDSliteCE, cls).tearDownClass()
295 def tearDownClass(cls):
296 super(TestDSliteCE, cls).tearDownClass()
298 def test_dslite_ce(self):
299 """Test DS-Lite CE"""
301 b4_ip6 = "2001:db8:62aa::375e:f4c1:1"
302 self.vapi.dslite_set_b4_addr(ip4_addr=b4_ip4, ip6_addr=b4_ip6)
304 aftr_ip4 = "192.0.0.1"
305 aftr_ip6 = "2001:db8:85a3::8a2e:370:1"
306 aftr_ip6_n = socket.inet_pton(socket.AF_INET6, aftr_ip6)
307 self.vapi.dslite_set_aftr_addr(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
313 [VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
319 Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
320 / IP(dst=self.pg1.remote_ip4, src=self.pg0.remote_ip4)
321 / UDP(sport=10000, dport=20000)
323 self.pg0.add_stream(p)
324 self.pg_enable_capture(self.pg_interfaces)
326 capture = self.pg1.get_capture(1)
328 self.assertEqual(capture[IPv6].src, b4_ip6)
329 self.assertEqual(capture[IPv6].dst, aftr_ip6)
330 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
331 self.assertEqual(capture[IP].dst, self.pg1.remote_ip4)
332 self.assertEqual(capture[UDP].sport, 10000)
333 self.assertEqual(capture[UDP].dport, 20000)
334 self.assert_packet_checksums_valid(capture)
338 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
339 / IPv6(dst=b4_ip6, src=aftr_ip6)
340 / IP(dst=self.pg0.remote_ip4, src=self.pg1.remote_ip4)
341 / UDP(sport=20000, dport=10000)
343 self.pg1.add_stream(p)
344 self.pg_enable_capture(self.pg_interfaces)
346 capture = self.pg0.get_capture(1)
348 self.assertFalse(capture.haslayer(IPv6))
349 self.assertEqual(capture[IP].src, self.pg1.remote_ip4)
350 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
351 self.assertEqual(capture[UDP].sport, 20000)
352 self.assertEqual(capture[UDP].dport, 10000)
353 self.assert_packet_checksums_valid(capture)
355 # ping DS-Lite B4 tunnel endpoint address
357 Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
358 / IPv6(src=self.pg1.remote_hosts[0].ip6, dst=b4_ip6)
359 / ICMPv6EchoRequest()
361 self.pg1.add_stream(p)
362 self.pg_enable_capture(self.pg_interfaces)
364 capture = self.pg1.get_capture(1)
366 self.assertEqual(capture[IPv6].src, b4_ip6)
367 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
368 self.assertTrue(capture.haslayer(ICMPv6EchoReply))
371 super(TestDSliteCE, self).tearDown()
373 def show_commands_at_teardown(self):
374 self.logger.info(self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
375 self.logger.info(self.vapi.cli("show dslite b4-tunnel-endpoint-address"))