8 from framework import tag_fixme_vpp_workers
9 from framework import VppTestCase, VppTestRunner, running_extended_tests
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 IPv6, ICMPv6EchoRequest, ICMPv6EchoReply, \
15 ICMPv6ND_NS, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, fragment6
16 from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6, IPv6ExtHdrFragment
17 from scapy.layers.l2 import Ether, ARP, GRE
18 from scapy.data import IP_PROTOS
19 from scapy.packet import bind_layers, Raw
21 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
22 from time import sleep
23 from util import ip4_range
24 from vpp_papi import mac_pton
25 from syslog_rfc5424_parser import SyslogMessage, ParseError
26 from syslog_rfc5424_parser.constants import SyslogFacility, SyslogSeverity
27 from io import BytesIO
28 from vpp_papi import VppEnum
29 from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathType
30 from vpp_neighbor import VppNeighbor
31 from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
32 IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
34 from ipaddress import IPv6Network
37 @tag_fixme_vpp_workers
38 class TestDSlite(VppTestCase):
39 """ DS-Lite Test Cases """
43 super(TestDSlite, cls).setUpClass()
46 cls.nat_addr = '10.0.0.3'
48 cls.create_pg_interfaces(range(3))
54 cls.pg1.generate_remote_hosts(2)
55 cls.pg1.configure_ipv6_neighbors()
61 super(TestDSlite, cls).tearDownClass()
65 def tearDownClass(cls):
66 super(TestDSlite, cls).tearDownClass()
68 def verify_syslog_apmadd(self, data, isaddr, isport, xsaddr, xsport,
70 message = data.decode('utf-8')
72 message = SyslogMessage.parse(message)
73 except ParseError as e:
76 self.assertEqual(message.severity, SyslogSeverity.info)
77 self.assertEqual(message.appname, 'NAT')
78 self.assertEqual(message.msgid, 'APMADD')
79 sd_params = message.sd.get('napmap')
80 self.assertTrue(sd_params is not None)
81 self.assertEqual(sd_params.get('IATYP'), 'IPv4')
82 self.assertEqual(sd_params.get('ISADDR'), isaddr)
83 self.assertEqual(sd_params.get('ISPORT'), "%d" % isport)
84 self.assertEqual(sd_params.get('XATYP'), 'IPv4')
85 self.assertEqual(sd_params.get('XSADDR'), xsaddr)
86 self.assertEqual(sd_params.get('XSPORT'), "%d" % xsport)
87 self.assertEqual(sd_params.get('PROTO'), "%d" % proto)
88 self.assertTrue(sd_params.get('SSUBIX') is not None)
89 self.assertEqual(sd_params.get('SV6ENC'), sv6enc)
91 def test_dslite(self):
93 nat_config = self.vapi.nat_show_config()
94 self.assertEqual(0, nat_config.dslite_ce)
96 self.vapi.dslite_add_del_pool_addr_range(start_addr=self.nat_addr,
97 end_addr=self.nat_addr,
99 aftr_ip4 = '192.0.0.1'
100 aftr_ip6 = '2001:db8:85a3::8a2e:370:1'
101 self.vapi.dslite_set_aftr_addr(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
102 self.vapi.syslog_set_sender(self.pg2.local_ip4, self.pg2.remote_ip4)
105 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
106 IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[0].ip6) /
107 IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
108 UDP(sport=20000, dport=10000))
109 self.pg1.add_stream(p)
110 self.pg_enable_capture(self.pg_interfaces)
112 capture = self.pg0.get_capture(1)
114 self.assertFalse(capture.haslayer(IPv6))
115 self.assertEqual(capture[IP].src, self.nat_addr)
116 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
117 self.assertNotEqual(capture[UDP].sport, 20000)
118 self.assertEqual(capture[UDP].dport, 10000)
119 self.assert_packet_checksums_valid(capture)
120 out_port = capture[UDP].sport
121 capture = self.pg2.get_capture(1)
122 self.verify_syslog_apmadd(capture[0][Raw].load, '192.168.1.1',
123 20000, self.nat_addr, out_port,
124 self.pg1.remote_hosts[0].ip6, IP_PROTOS.udp)
126 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
127 IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
128 UDP(sport=10000, dport=out_port))
129 self.pg0.add_stream(p)
130 self.pg_enable_capture(self.pg_interfaces)
132 capture = self.pg1.get_capture(1)
134 self.assertEqual(capture[IPv6].src, aftr_ip6)
135 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
136 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
137 self.assertEqual(capture[IP].dst, '192.168.1.1')
138 self.assertEqual(capture[UDP].sport, 10000)
139 self.assertEqual(capture[UDP].dport, 20000)
140 self.assert_packet_checksums_valid(capture)
143 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
144 IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
145 IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
146 TCP(sport=20001, dport=10001))
147 self.pg1.add_stream(p)
148 self.pg_enable_capture(self.pg_interfaces)
150 capture = self.pg0.get_capture(1)
152 self.assertFalse(capture.haslayer(IPv6))
153 self.assertEqual(capture[IP].src, self.nat_addr)
154 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
155 self.assertNotEqual(capture[TCP].sport, 20001)
156 self.assertEqual(capture[TCP].dport, 10001)
157 self.assert_packet_checksums_valid(capture)
158 out_port = capture[TCP].sport
160 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
161 IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
162 TCP(sport=10001, dport=out_port))
163 self.pg0.add_stream(p)
164 self.pg_enable_capture(self.pg_interfaces)
166 capture = self.pg1.get_capture(1)
168 self.assertEqual(capture[IPv6].src, aftr_ip6)
169 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
170 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
171 self.assertEqual(capture[IP].dst, '192.168.1.1')
172 self.assertEqual(capture[TCP].sport, 10001)
173 self.assertEqual(capture[TCP].dport, 20001)
174 self.assert_packet_checksums_valid(capture)
177 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
178 IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
179 IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
180 ICMP(id=4000, type='echo-request'))
181 self.pg1.add_stream(p)
182 self.pg_enable_capture(self.pg_interfaces)
184 capture = self.pg0.get_capture(1)
186 self.assertFalse(capture.haslayer(IPv6))
187 self.assertEqual(capture[IP].src, self.nat_addr)
188 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
189 self.assertNotEqual(capture[ICMP].id, 4000)
190 self.assert_packet_checksums_valid(capture)
191 out_id = capture[ICMP].id
193 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
194 IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
195 ICMP(id=out_id, type='echo-reply'))
196 self.pg0.add_stream(p)
197 self.pg_enable_capture(self.pg_interfaces)
199 capture = self.pg1.get_capture(1)
201 self.assertEqual(capture[IPv6].src, aftr_ip6)
202 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
203 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
204 self.assertEqual(capture[IP].dst, '192.168.1.1')
205 self.assertEqual(capture[ICMP].id, 4000)
206 self.assert_packet_checksums_valid(capture)
208 # ping DS-Lite AFTR tunnel endpoint address
209 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
210 IPv6(src=self.pg1.remote_hosts[1].ip6, dst=aftr_ip6) /
212 self.pg1.add_stream(p)
213 self.pg_enable_capture(self.pg_interfaces)
215 capture = self.pg1.get_capture(1)
217 self.assertEqual(capture[IPv6].src, aftr_ip6)
218 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
219 self.assertTrue(capture.haslayer(ICMPv6EchoReply))
221 b4s = self.statistics.get_counter('/dslite/total-b4s')
222 self.assertEqual(b4s[0][0], 2)
223 sessions = self.statistics.get_counter('/dslite/total-sessions')
224 self.assertEqual(sessions[0][0], 3)
227 super(TestDSlite, self).tearDown()
229 def show_commands_at_teardown(self):
230 self.logger.info(self.vapi.cli("show dslite pool"))
232 self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
233 self.logger.info(self.vapi.cli("show dslite sessions"))
236 class TestDSliteCE(VppTestCase):
237 """ DS-Lite CE Test Cases """
240 def setUpConstants(cls):
241 super(TestDSliteCE, cls).setUpConstants()
242 cls.vpp_cmdline.extend(["dslite", "{", "ce", "}"])
246 super(TestDSliteCE, cls).setUpClass()
249 cls.create_pg_interfaces(range(2))
252 cls.pg0.resolve_arp()
255 cls.pg1.generate_remote_hosts(1)
256 cls.pg1.configure_ipv6_neighbors()
259 super(TestDSliteCE, cls).tearDownClass()
263 def tearDownClass(cls):
264 super(TestDSliteCE, cls).tearDownClass()
266 def test_dslite_ce(self):
267 """ Test DS-Lite CE """
269 # TODO: add message to retrieve dslite config
270 # nat_config = self.vapi.nat_show_config()
271 # self.assertEqual(1, nat_config.dslite_ce)
274 b4_ip6 = '2001:db8:62aa::375e:f4c1:1'
275 self.vapi.dslite_set_b4_addr(ip4_addr=b4_ip4, ip6_addr=b4_ip6)
277 aftr_ip4 = '192.0.0.1'
278 aftr_ip6 = '2001:db8:85a3::8a2e:370:1'
279 aftr_ip6_n = socket.inet_pton(socket.AF_INET6, aftr_ip6)
280 self.vapi.dslite_set_aftr_addr(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
282 r1 = VppIpRoute(self, aftr_ip6, 128,
283 [VppRoutePath(self.pg1.remote_ip6,
284 self.pg1.sw_if_index)])
288 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
289 IP(dst=self.pg1.remote_ip4, src=self.pg0.remote_ip4) /
290 UDP(sport=10000, dport=20000))
291 self.pg0.add_stream(p)
292 self.pg_enable_capture(self.pg_interfaces)
294 capture = self.pg1.get_capture(1)
296 self.assertEqual(capture[IPv6].src, b4_ip6)
297 self.assertEqual(capture[IPv6].dst, aftr_ip6)
298 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
299 self.assertEqual(capture[IP].dst, self.pg1.remote_ip4)
300 self.assertEqual(capture[UDP].sport, 10000)
301 self.assertEqual(capture[UDP].dport, 20000)
302 self.assert_packet_checksums_valid(capture)
305 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
306 IPv6(dst=b4_ip6, src=aftr_ip6) /
307 IP(dst=self.pg0.remote_ip4, src=self.pg1.remote_ip4) /
308 UDP(sport=20000, dport=10000))
309 self.pg1.add_stream(p)
310 self.pg_enable_capture(self.pg_interfaces)
312 capture = self.pg0.get_capture(1)
314 self.assertFalse(capture.haslayer(IPv6))
315 self.assertEqual(capture[IP].src, self.pg1.remote_ip4)
316 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
317 self.assertEqual(capture[UDP].sport, 20000)
318 self.assertEqual(capture[UDP].dport, 10000)
319 self.assert_packet_checksums_valid(capture)
321 # ping DS-Lite B4 tunnel endpoint address
322 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
323 IPv6(src=self.pg1.remote_hosts[0].ip6, dst=b4_ip6) /
325 self.pg1.add_stream(p)
326 self.pg_enable_capture(self.pg_interfaces)
328 capture = self.pg1.get_capture(1)
330 self.assertEqual(capture[IPv6].src, b4_ip6)
331 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
332 self.assertTrue(capture.haslayer(ICMPv6EchoReply))
335 super(TestDSliteCE, self).tearDown()
337 def show_commands_at_teardown(self):
339 self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
341 self.vapi.cli("show dslite b4-tunnel-endpoint-address"))