8 from framework import VppTestCase, VppTestRunner, running_extended_tests
11 from scapy.layers.inet import IP, TCP, UDP, ICMP
12 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
13 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply, \
14 ICMPv6ND_NS, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, fragment6
15 from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6, IPv6ExtHdrFragment
16 from scapy.layers.l2 import Ether, ARP, GRE
17 from scapy.data import IP_PROTOS
18 from scapy.packet import bind_layers, Raw
20 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
21 from time import sleep
22 from util import ip4_range
23 from vpp_papi import mac_pton
24 from syslog_rfc5424_parser import SyslogMessage, ParseError
25 from syslog_rfc5424_parser.constants import SyslogFacility, SyslogSeverity
26 from io import BytesIO
27 from vpp_papi import VppEnum
28 from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathType
29 from vpp_neighbor import VppNeighbor
30 from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
31 IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
33 from ipaddress import IPv6Network
36 class TestDSlite(VppTestCase):
37 """ DS-Lite Test Cases """
41 super(TestDSlite, cls).setUpClass()
44 cls.nat_addr = '10.0.0.3'
46 cls.create_pg_interfaces(range(3))
52 cls.pg1.generate_remote_hosts(2)
53 cls.pg1.configure_ipv6_neighbors()
59 super(TestDSlite, cls).tearDownClass()
63 def tearDownClass(cls):
64 super(TestDSlite, cls).tearDownClass()
66 def verify_syslog_apmadd(self, data, isaddr, isport, xsaddr, xsport,
68 message = data.decode('utf-8')
70 message = SyslogMessage.parse(message)
71 except ParseError as e:
74 self.assertEqual(message.severity, SyslogSeverity.info)
75 self.assertEqual(message.appname, 'NAT')
76 self.assertEqual(message.msgid, 'APMADD')
77 sd_params = message.sd.get('napmap')
78 self.assertTrue(sd_params is not None)
79 self.assertEqual(sd_params.get('IATYP'), 'IPv4')
80 self.assertEqual(sd_params.get('ISADDR'), isaddr)
81 self.assertEqual(sd_params.get('ISPORT'), "%d" % isport)
82 self.assertEqual(sd_params.get('XATYP'), 'IPv4')
83 self.assertEqual(sd_params.get('XSADDR'), xsaddr)
84 self.assertEqual(sd_params.get('XSPORT'), "%d" % xsport)
85 self.assertEqual(sd_params.get('PROTO'), "%d" % proto)
86 self.assertTrue(sd_params.get('SSUBIX') is not None)
87 self.assertEqual(sd_params.get('SV6ENC'), sv6enc)
89 def test_dslite(self):
91 nat_config = self.vapi.nat_show_config()
92 self.assertEqual(0, nat_config.dslite_ce)
94 self.vapi.dslite_add_del_pool_addr_range(start_addr=self.nat_addr,
95 end_addr=self.nat_addr,
97 aftr_ip4 = '192.0.0.1'
98 aftr_ip6 = '2001:db8:85a3::8a2e:370:1'
99 self.vapi.dslite_set_aftr_addr(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
100 self.vapi.syslog_set_sender(self.pg2.local_ip4, self.pg2.remote_ip4)
103 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
104 IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[0].ip6) /
105 IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
106 UDP(sport=20000, dport=10000))
107 self.pg1.add_stream(p)
108 self.pg_enable_capture(self.pg_interfaces)
110 capture = self.pg0.get_capture(1)
112 self.assertFalse(capture.haslayer(IPv6))
113 self.assertEqual(capture[IP].src, self.nat_addr)
114 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
115 self.assertNotEqual(capture[UDP].sport, 20000)
116 self.assertEqual(capture[UDP].dport, 10000)
117 self.assert_packet_checksums_valid(capture)
118 out_port = capture[UDP].sport
119 capture = self.pg2.get_capture(1)
120 self.verify_syslog_apmadd(capture[0][Raw].load, '192.168.1.1',
121 20000, self.nat_addr, out_port,
122 self.pg1.remote_hosts[0].ip6, IP_PROTOS.udp)
124 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
125 IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
126 UDP(sport=10000, dport=out_port))
127 self.pg0.add_stream(p)
128 self.pg_enable_capture(self.pg_interfaces)
130 capture = self.pg1.get_capture(1)
132 self.assertEqual(capture[IPv6].src, aftr_ip6)
133 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
134 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
135 self.assertEqual(capture[IP].dst, '192.168.1.1')
136 self.assertEqual(capture[UDP].sport, 10000)
137 self.assertEqual(capture[UDP].dport, 20000)
138 self.assert_packet_checksums_valid(capture)
141 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
142 IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
143 IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
144 TCP(sport=20001, dport=10001))
145 self.pg1.add_stream(p)
146 self.pg_enable_capture(self.pg_interfaces)
148 capture = self.pg0.get_capture(1)
150 self.assertFalse(capture.haslayer(IPv6))
151 self.assertEqual(capture[IP].src, self.nat_addr)
152 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
153 self.assertNotEqual(capture[TCP].sport, 20001)
154 self.assertEqual(capture[TCP].dport, 10001)
155 self.assert_packet_checksums_valid(capture)
156 out_port = capture[TCP].sport
158 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
159 IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
160 TCP(sport=10001, dport=out_port))
161 self.pg0.add_stream(p)
162 self.pg_enable_capture(self.pg_interfaces)
164 capture = self.pg1.get_capture(1)
166 self.assertEqual(capture[IPv6].src, aftr_ip6)
167 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
168 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
169 self.assertEqual(capture[IP].dst, '192.168.1.1')
170 self.assertEqual(capture[TCP].sport, 10001)
171 self.assertEqual(capture[TCP].dport, 20001)
172 self.assert_packet_checksums_valid(capture)
175 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
176 IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
177 IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
178 ICMP(id=4000, type='echo-request'))
179 self.pg1.add_stream(p)
180 self.pg_enable_capture(self.pg_interfaces)
182 capture = self.pg0.get_capture(1)
184 self.assertFalse(capture.haslayer(IPv6))
185 self.assertEqual(capture[IP].src, self.nat_addr)
186 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
187 self.assertNotEqual(capture[ICMP].id, 4000)
188 self.assert_packet_checksums_valid(capture)
189 out_id = capture[ICMP].id
191 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
192 IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
193 ICMP(id=out_id, type='echo-reply'))
194 self.pg0.add_stream(p)
195 self.pg_enable_capture(self.pg_interfaces)
197 capture = self.pg1.get_capture(1)
199 self.assertEqual(capture[IPv6].src, aftr_ip6)
200 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
201 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
202 self.assertEqual(capture[IP].dst, '192.168.1.1')
203 self.assertEqual(capture[ICMP].id, 4000)
204 self.assert_packet_checksums_valid(capture)
206 # ping DS-Lite AFTR tunnel endpoint address
207 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
208 IPv6(src=self.pg1.remote_hosts[1].ip6, dst=aftr_ip6) /
210 self.pg1.add_stream(p)
211 self.pg_enable_capture(self.pg_interfaces)
213 capture = self.pg1.get_capture(1)
215 self.assertEqual(capture[IPv6].src, aftr_ip6)
216 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
217 self.assertTrue(capture.haslayer(ICMPv6EchoReply))
219 b4s = self.statistics.get_counter('/dslite/total-b4s')
220 self.assertEqual(b4s[0][0], 2)
221 sessions = self.statistics.get_counter('/dslite/total-sessions')
222 self.assertEqual(sessions[0][0], 3)
225 super(TestDSlite, self).tearDown()
227 def show_commands_at_teardown(self):
228 self.logger.info(self.vapi.cli("show dslite pool"))
230 self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
231 self.logger.info(self.vapi.cli("show dslite sessions"))
234 class TestDSliteCE():
235 """ DS-Lite CE Test Cases """
238 def setUpConstants(cls):
239 super(TestDSliteCE, cls).setUpConstants()
240 cls.vpp_cmdline.extend(["nat", "{", "dslite ce", "}"])
244 super(TestDSliteCE, cls).setUpClass()
247 cls.create_pg_interfaces(range(2))
250 cls.pg0.resolve_arp()
253 cls.pg1.generate_remote_hosts(1)
254 cls.pg1.configure_ipv6_neighbors()
257 super(TestDSliteCE, cls).tearDownClass()
261 def tearDownClass(cls):
262 super(TestDSliteCE, cls).tearDownClass()
264 def test_dslite_ce(self):
265 """ Test DS-Lite CE """
267 nat_config = self.vapi.nat_show_config()
268 self.assertEqual(1, nat_config.dslite_ce)
271 b4_ip6 = '2001:db8:62aa::375e:f4c1:1'
272 self.vapi.dslite_set_b4_addr(ip4_addr=b4_ip4, ip6_addr=b4_ip6)
274 aftr_ip4 = '192.0.0.1'
275 aftr_ip6 = '2001:db8:85a3::8a2e:370:1'
276 aftr_ip6_n = socket.inet_pton(socket.AF_INET6, aftr_ip6)
277 self.vapi.dslite_set_aftr_addr(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
279 r1 = VppIpRoute(self, aftr_ip6, 128,
280 [VppRoutePath(self.pg1.remote_ip6,
281 self.pg1.sw_if_index)])
285 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
286 IP(dst=self.pg1.remote_ip4, src=self.pg0.remote_ip4) /
287 UDP(sport=10000, dport=20000))
288 self.pg0.add_stream(p)
289 self.pg_enable_capture(self.pg_interfaces)
291 capture = self.pg1.get_capture(1)
293 self.assertEqual(capture[IPv6].src, b4_ip6)
294 self.assertEqual(capture[IPv6].dst, aftr_ip6)
295 self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
296 self.assertEqual(capture[IP].dst, self.pg1.remote_ip4)
297 self.assertEqual(capture[UDP].sport, 10000)
298 self.assertEqual(capture[UDP].dport, 20000)
299 self.assert_packet_checksums_valid(capture)
302 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
303 IPv6(dst=b4_ip6, src=aftr_ip6) /
304 IP(dst=self.pg0.remote_ip4, src=self.pg1.remote_ip4) /
305 UDP(sport=20000, dport=10000))
306 self.pg1.add_stream(p)
307 self.pg_enable_capture(self.pg_interfaces)
309 capture = self.pg0.get_capture(1)
311 self.assertFalse(capture.haslayer(IPv6))
312 self.assertEqual(capture[IP].src, self.pg1.remote_ip4)
313 self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
314 self.assertEqual(capture[UDP].sport, 20000)
315 self.assertEqual(capture[UDP].dport, 10000)
316 self.assert_packet_checksums_valid(capture)
318 # ping DS-Lite B4 tunnel endpoint address
319 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
320 IPv6(src=self.pg1.remote_hosts[0].ip6, dst=b4_ip6) /
322 self.pg1.add_stream(p)
323 self.pg_enable_capture(self.pg_interfaces)
325 capture = self.pg1.get_capture(1)
327 self.assertEqual(capture[IPv6].src, b4_ip6)
328 self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
329 self.assertTrue(capture.haslayer(ICMPv6EchoReply))
332 super(TestDSliteCE, self).tearDown()
334 def show_commands_at_teardown(self):
336 self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
338 self.vapi.cli("show dslite b4-tunnel-endpoint-address"))