src/plugins/map/test/test_map.py

   1 #!/usr/bin/env python3
   2
   3 import ipaddress
   4 import unittest
   5
   6 from framework import VppTestCase, VppTestRunner
   7 from vpp_ip import DpoProto
   8 from vpp_ip_route import VppIpRoute, VppRoutePath
   9 from util import fragment_rfc791, fragment_rfc8200
  10
  11 import scapy.compat
  12 from scapy.layers.l2 import Ether
  13 from scapy.packet import Raw
  14 from scapy.layers.inet import IP, UDP, ICMP, TCP
  15 from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded, IPv6ExtHdrFragment, \
  16     ICMPv6EchoRequest, ICMPv6DestUnreach
  17
  18
  19 class TestMAP(VppTestCase):
  20     """ MAP Test Case """
  21
  22     @classmethod
  23     def setUpClass(cls):
  24         super(TestMAP, cls).setUpClass()
  25
  26     @classmethod
  27     def tearDownClass(cls):
  28         super(TestMAP, cls).tearDownClass()
  29
  30     def setUp(self):
  31         super(TestMAP, self).setUp()
  32
  33         # create 2 pg interfaces
  34         self.create_pg_interfaces(range(4))
  35
  36         # pg0 is 'inside' IPv4
  37         self.pg0.admin_up()
  38         self.pg0.config_ip4()
  39         self.pg0.resolve_arp()
  40         self.pg0.generate_remote_hosts(2)
  41         self.pg0.configure_ipv4_neighbors()
  42
  43         # pg1 is 'outside' IPv6
  44         self.pg1.admin_up()
  45         self.pg1.config_ip6()
  46         self.pg1.generate_remote_hosts(4)
  47         self.pg1.configure_ipv6_neighbors()
  48
  49     def tearDown(self):
  50         super(TestMAP, self).tearDown()
  51         for i in self.pg_interfaces:
  52             i.unconfig_ip4()
  53             i.unconfig_ip6()
  54             i.admin_down()
  55
  56     def send_and_assert_encapped(self, packets, ip6_src, ip6_dst, dmac=None):
  57         if not dmac:
  58             dmac = self.pg1.remote_mac
  59
  60         self.pg0.add_stream(packets)
  61
  62         self.pg_enable_capture(self.pg_interfaces)
  63         self.pg_start()
  64
  65         capture = self.pg1.get_capture(len(packets))
  66         for rx, tx in zip(capture, packets):
  67             self.assertEqual(rx[Ether].dst, dmac)
  68             self.assertEqual(rx[IP].src, tx[IP].src)
  69             self.assertEqual(rx[IPv6].src, ip6_src)
  70             self.assertEqual(rx[IPv6].dst, ip6_dst)
  71
  72     def send_and_assert_encapped_one(self, packet, ip6_src, ip6_dst,
  73                                      dmac=None):
  74         return self.send_and_assert_encapped([packet], ip6_src, ip6_dst, dmac)
  75
  76     def test_api_map_domain_dump(self):
  77         map_dst = '2001::/64'
  78         map_src = '3000::1/128'
  79         client_pfx = '192.168.0.0/16'
  80         tag = 'MAP-E tag.'
  81         index = self.vapi.map_add_domain(ip4_prefix=client_pfx,
  82                                          ip6_prefix=map_dst,
  83                                          ip6_src=map_src,
  84                                          tag=tag).index
  85         rv = self.vapi.map_domain_dump()
  86
  87         # restore the state early so as to not impact subsequent tests.
  88         # If an assert fails, we will not get the chance to do it at the end.
  89         self.vapi.map_del_domain(index=index)
  90
  91         self.assertGreater(len(rv), 0,
  92                            "Expected output from 'map_domain_dump'")
  93
  94         # typedefs are returned as ipaddress objects.
  95         # wrap results in str() ugh! to avoid the need to call unicode.
  96         self.assertEqual(str(rv[0].ip4_prefix), client_pfx)
  97         self.assertEqual(str(rv[0].ip6_prefix), map_dst)
  98         self.assertEqual(str(rv[0].ip6_src), map_src)
  99
 100         self.assertEqual(rv[0].tag, tag,
 101                          "output produced incorrect tag value.")
 102
 103     def test_map_e_udp(self):
 104         """ MAP-E UDP"""
 105
 106         #
 107         # Add a route to the MAP-BR
 108         #
 109         map_br_pfx = "2001::"
 110         map_br_pfx_len = 32
 111         map_route = VppIpRoute(self,
 112                                map_br_pfx,
 113                                map_br_pfx_len,
 114                                [VppRoutePath(self.pg1.remote_ip6,
 115                                              self.pg1.sw_if_index)])
 116         map_route.add_vpp_config()
 117
 118         #
 119         # Add a domain that maps from pg0 to pg1
 120         #
 121         map_dst = '2001::/32'
 122         map_src = '3000::1/128'
 123         client_pfx = '192.168.0.0/16'
 124         map_translated_addr = '2001:0:101:7000:0:c0a8:101:7'
 125         tag = 'MAP-E tag.'
 126         self.vapi.map_add_domain(ip4_prefix=client_pfx,
 127                                  ip6_prefix=map_dst,
 128                                  ip6_src=map_src,
 129                                  ea_bits_len=20,
 130                                  psid_offset=4,
 131                                  psid_length=4,
 132                                  tag=tag)
 133
 134         self.vapi.map_param_set_security_check(enable=1, fragments=1)
 135
 136         # Enable MAP on interface.
 137         self.vapi.map_if_enable_disable(is_enable=1,
 138                                         sw_if_index=self.pg0.sw_if_index,
 139                                         is_translation=0)
 140
 141         # Ensure MAP doesn't steal all packets!
 142         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 143               IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
 144               UDP(sport=20000, dport=10000) /
 145               Raw(b'\xa5' * 100))
 146         rx = self.send_and_expect(self.pg0, v4 * 4, self.pg0)
 147         v4_reply = v4[1]
 148         v4_reply.ttl -= 1
 149         for p in rx:
 150             self.validate(p[1], v4_reply)
 151
 152         #
 153         # Fire in a v4 packet that will be encapped to the BR
 154         #
 155         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 156               IP(src=self.pg0.remote_ip4, dst='192.168.1.1') /
 157               UDP(sport=20000, dport=10000) /
 158               Raw(b'\xa5' * 100))
 159
 160         self.send_and_assert_encapped(v4 * 4, "3000::1", map_translated_addr)
 161
 162         #
 163         # Verify reordered fragments are able to pass as well
 164         #
 165         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 166               IP(id=1, src=self.pg0.remote_ip4, dst='192.168.1.1') /
 167               UDP(sport=20000, dport=10000) /
 168               Raw(b'\xa5' * 1000))
 169
 170         frags = fragment_rfc791(v4, 400)
 171         frags.reverse()
 172
 173         self.send_and_assert_encapped(frags, "3000::1", map_translated_addr)
 174
 175         # Enable MAP on interface.
 176         self.vapi.map_if_enable_disable(is_enable=1,
 177                                         sw_if_index=self.pg1.sw_if_index,
 178                                         is_translation=0)
 179
 180         # Ensure MAP doesn't steal all packets
 181         v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 182               IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
 183               UDP(sport=20000, dport=10000) /
 184               Raw(b'\xa5' * 100))
 185         rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
 186         v6_reply = v6[1]
 187         v6_reply.hlim -= 1
 188         for p in rx:
 189             self.validate(p[1], v6_reply)
 190
 191         #
 192         # Fire in a V6 encapped packet.
 193         # expect a decapped packet on the inside ip4 link
 194         #
 195         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 196              IPv6(dst='3000::1', src=map_translated_addr) /
 197              IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
 198              UDP(sport=10000, dport=20000) /
 199              Raw(b'\xa5' * 100))
 200
 201         self.pg1.add_stream(p)
 202
 203         self.pg_enable_capture(self.pg_interfaces)
 204         self.pg_start()
 205
 206         rx = self.pg0.get_capture(1)
 207         rx = rx[0]
 208
 209         self.assertFalse(rx.haslayer(IPv6))
 210         self.assertEqual(rx[IP].src, p[IP].src)
 211         self.assertEqual(rx[IP].dst, p[IP].dst)
 212
 213         #
 214         # Verify encapped reordered fragments pass as well
 215         #
 216         p = (IP(id=1, dst=self.pg0.remote_ip4, src='192.168.1.1') /
 217              UDP(sport=10000, dport=20000) /
 218              Raw(b'\xa5' * 1500))
 219         frags = fragment_rfc791(p, 400)
 220         frags.reverse()
 221
 222         stream = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 223                   IPv6(dst='3000::1', src=map_translated_addr) /
 224                   x for x in frags)
 225
 226         self.pg1.add_stream(stream)
 227
 228         self.pg_enable_capture(self.pg_interfaces)
 229         self.pg_start()
 230
 231         rx = self.pg0.get_capture(len(frags))
 232
 233         for r in rx:
 234             self.assertFalse(r.haslayer(IPv6))
 235             self.assertEqual(r[IP].src, p[IP].src)
 236             self.assertEqual(r[IP].dst, p[IP].dst)
 237
 238         # Verify that fragments pass even if ipv6 layer is fragmented
 239         stream = (IPv6(dst='3000::1', src=map_translated_addr) / x
 240                   for x in frags)
 241
 242         v6_stream = [
 243             Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / x
 244             for i in range(len(frags))
 245             for x in fragment_rfc8200(
 246                 IPv6(dst='3000::1', src=map_translated_addr) / frags[i],
 247                 i, 200)]
 248
 249         self.pg1.add_stream(v6_stream)
 250
 251         self.pg_enable_capture(self.pg_interfaces)
 252         self.pg_start()
 253
 254         rx = self.pg0.get_capture(len(frags))
 255
 256         for r in rx:
 257             self.assertFalse(r.haslayer(IPv6))
 258             self.assertEqual(r[IP].src, p[IP].src)
 259             self.assertEqual(r[IP].dst, p[IP].dst)
 260
 261         #
 262         # Pre-resolve. No API for this!!
 263         #
 264         self.vapi.ppcli("map params pre-resolve ip6-nh 4001::1")
 265
 266         self.send_and_assert_no_replies(self.pg0, v4,
 267                                         "resolved via default route")
 268
 269         #
 270         # Add a route to 4001::1. Expect the encapped traffic to be
 271         # sent via that routes next-hop
 272         #
 273         pre_res_route = VppIpRoute(self, "4001::1", 128,
 274                                    [VppRoutePath(self.pg1.remote_hosts[2].ip6,
 275                                                  self.pg1.sw_if_index)])
 276         pre_res_route.add_vpp_config()
 277
 278         self.send_and_assert_encapped_one(v4, "3000::1",
 279                                           map_translated_addr,
 280                                           dmac=self.pg1.remote_hosts[2].mac)
 281
 282         #
 283         # change the route to the pre-solved next-hop
 284         #
 285         pre_res_route.modify([VppRoutePath(self.pg1.remote_hosts[3].ip6,
 286                                            self.pg1.sw_if_index)])
 287         pre_res_route.add_vpp_config()
 288
 289         self.send_and_assert_encapped_one(v4, "3000::1",
 290                                           map_translated_addr,
 291                                           dmac=self.pg1.remote_hosts[3].mac)
 292
 293         #
 294         # cleanup. The test infra's object registry will ensure
 295         # the route is really gone and thus that the unresolve worked.
 296         #
 297         pre_res_route.remove_vpp_config()
 298         self.vapi.ppcli("map params pre-resolve del ip6-nh 4001::1")
 299
 300     def test_map_e_inner_frag(self):
 301         """ MAP-E Inner fragmentation """
 302
 303         #
 304         # Add a route to the MAP-BR
 305         #
 306         map_br_pfx = "2001::"
 307         map_br_pfx_len = 32
 308         map_route = VppIpRoute(self,
 309                                map_br_pfx,
 310                                map_br_pfx_len,
 311                                [VppRoutePath(self.pg1.remote_ip6,
 312                                              self.pg1.sw_if_index)])
 313         map_route.add_vpp_config()
 314
 315         #
 316         # Add a domain that maps from pg0 to pg1
 317         #
 318         map_dst = '2001::/32'
 319         map_src = '3000::1/128'
 320         client_pfx = '192.168.0.0/16'
 321         map_translated_addr = '2001:0:101:7000:0:c0a8:101:7'
 322         tag = 'MAP-E tag.'
 323         self.vapi.map_add_domain(ip4_prefix=client_pfx,
 324                                  ip6_prefix=map_dst,
 325                                  ip6_src=map_src,
 326                                  ea_bits_len=20,
 327                                  psid_offset=4,
 328                                  psid_length=4,
 329                                  mtu=1000,
 330                                  tag=tag)
 331
 332         # Enable MAP on interface.
 333         self.vapi.map_if_enable_disable(is_enable=1,
 334                                         sw_if_index=self.pg0.sw_if_index,
 335                                         is_translation=0)
 336
 337         # Enable inner fragmentation
 338         self.vapi.map_param_set_fragmentation(inner=1)
 339
 340         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 341               IP(src=self.pg0.remote_ip4, dst='192.168.1.1') /
 342               UDP(sport=20000, dport=10000) /
 343               Raw(b'\xa5' * 1300))
 344
 345         self.pg_send(self.pg0, v4*1)
 346         rx = self.pg1.get_capture(2)
 347
 348         frags = fragment_rfc791(v4[1], 1000)
 349         frags[0].id = 0
 350         frags[1].id = 0
 351         frags[0].ttl -= 1
 352         frags[1].ttl -= 1
 353         frags[0].chksum = 0
 354         frags[1].chksum = 0
 355
 356         v6_reply1 = (IPv6(src='3000::1', dst=map_translated_addr, hlim=63) /
 357                      frags[0])
 358         v6_reply2 = (IPv6(src='3000::1', dst=map_translated_addr, hlim=63) /
 359                      frags[1])
 360         rx[0][1].fl = 0
 361         rx[1][1].fl = 0
 362         rx[0][1][IP].id = 0
 363         rx[1][1][IP].id = 0
 364         rx[0][1][IP].chksum = 0
 365         rx[1][1][IP].chksum = 0
 366
 367         self.validate(rx[0][1], v6_reply1)
 368         self.validate(rx[1][1], v6_reply2)
 369
 370     def test_map_e_tcp_mss(self):
 371         """ MAP-E TCP MSS"""
 372
 373         #
 374         # Add a route to the MAP-BR
 375         #
 376         map_br_pfx = "2001::"
 377         map_br_pfx_len = 32
 378         map_route = VppIpRoute(self,
 379                                map_br_pfx,
 380                                map_br_pfx_len,
 381                                [VppRoutePath(self.pg1.remote_ip6,
 382                                              self.pg1.sw_if_index)])
 383         map_route.add_vpp_config()
 384
 385         #
 386         # Add a domain that maps from pg0 to pg1
 387         #
 388         map_dst = '2001::/32'
 389         map_src = '3000::1/128'
 390         client_pfx = '192.168.0.0/16'
 391         map_translated_addr = '2001:0:101:5000:0:c0a8:101:5'
 392         tag = 'MAP-E TCP tag.'
 393         self.vapi.map_add_domain(ip4_prefix=client_pfx,
 394                                  ip6_prefix=map_dst,
 395                                  ip6_src=map_src,
 396                                  ea_bits_len=20,
 397                                  psid_offset=4,
 398                                  psid_length=4,
 399                                  tag=tag)
 400
 401         # Enable MAP on pg0 interface.
 402         self.vapi.map_if_enable_disable(is_enable=1,
 403                                         sw_if_index=self.pg0.sw_if_index,
 404                                         is_translation=0)
 405
 406         # Enable MAP on pg1 interface.
 407         self.vapi.map_if_enable_disable(is_enable=1,
 408                                         sw_if_index=self.pg1.sw_if_index,
 409                                         is_translation=0)
 410
 411         # TCP MSS clamping
 412         mss_clamp = 1300
 413         self.vapi.map_param_set_tcp(mss_clamp)
 414
 415         #
 416         # Send a v4 packet that will be encapped.
 417         #
 418         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 419         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.1.1')
 420         p_tcp = TCP(sport=20000, dport=30000, flags="S",
 421                     options=[("MSS", 1455)])
 422         p4 = p_ether / p_ip4 / p_tcp
 423
 424         self.pg1.add_stream(p4)
 425         self.pg_enable_capture(self.pg_interfaces)
 426         self.pg_start()
 427
 428         rx = self.pg1.get_capture(1)
 429         rx = rx[0]
 430
 431         self.assertTrue(rx.haslayer(IPv6))
 432         self.assertEqual(rx[IP].src, p4[IP].src)
 433         self.assertEqual(rx[IP].dst, p4[IP].dst)
 434         self.assertEqual(rx[IPv6].src, "3000::1")
 435         self.assertEqual(rx[TCP].options,
 436                          TCP(options=[('MSS', mss_clamp)]).options)
 437
 438     def validate(self, rx, expected):
 439         self.assertEqual(rx, expected.__class__(scapy.compat.raw(expected)))
 440
 441     def validate_frag6(self, p6_frag, p_ip6_expected):
 442         self.assertFalse(p6_frag.haslayer(IP))
 443         self.assertTrue(p6_frag.haslayer(IPv6))
 444         self.assertTrue(p6_frag.haslayer(IPv6ExtHdrFragment))
 445         self.assertEqual(p6_frag[IPv6].src, p_ip6_expected.src)
 446         self.assertEqual(p6_frag[IPv6].dst, p_ip6_expected.dst)
 447
 448     def validate_frag_payload_len6(self, rx, proto, payload_len_expected):
 449         payload_total = 0
 450         for p in rx:
 451             payload_total += p[IPv6].plen
 452
 453         # First fragment has proto
 454         payload_total -= len(proto())
 455
 456         # Every fragment has IPv6 fragment header
 457         payload_total -= len(IPv6ExtHdrFragment()) * len(rx)
 458
 459         self.assertEqual(payload_total, payload_len_expected)
 460
 461     def validate_frag4(self, p4_frag, p_ip4_expected):
 462         self.assertFalse(p4_frag.haslayer(IPv6))
 463         self.assertTrue(p4_frag.haslayer(IP))
 464         self.assertTrue(p4_frag[IP].frag != 0 or p4_frag[IP].flags.MF)
 465         self.assertEqual(p4_frag[IP].src, p_ip4_expected.src)
 466         self.assertEqual(p4_frag[IP].dst, p_ip4_expected.dst)
 467
 468     def validate_frag_payload_len4(self, rx, proto, payload_len_expected):
 469         payload_total = 0
 470         for p in rx:
 471             payload_total += len(p[IP].payload)
 472
 473         # First fragment has proto
 474         payload_total -= len(proto())
 475
 476         self.assertEqual(payload_total, payload_len_expected)
 477
 478     def payload(self, len):
 479         return 'x' * len
 480
 481     def test_map_t(self):
 482         """ MAP-T """
 483
 484         #
 485         # Add a domain that maps from pg0 to pg1
 486         #
 487         map_dst = '2001:db8::/32'
 488         map_src = '1234:5678:90ab:cdef::/64'
 489         ip4_pfx = '192.168.0.0/24'
 490         tag = 'MAP-T Tag.'
 491
 492         self.vapi.map_add_domain(ip6_prefix=map_dst,
 493                                  ip4_prefix=ip4_pfx,
 494                                  ip6_src=map_src,
 495                                  ea_bits_len=16,
 496                                  psid_offset=6,
 497                                  psid_length=4,
 498                                  mtu=1500,
 499                                  tag=tag)
 500
 501         # Enable MAP-T on interfaces.
 502         self.vapi.map_if_enable_disable(is_enable=1,
 503                                         sw_if_index=self.pg0.sw_if_index,
 504                                         is_translation=1)
 505         self.vapi.map_if_enable_disable(is_enable=1,
 506                                         sw_if_index=self.pg1.sw_if_index,
 507                                         is_translation=1)
 508
 509         # Ensure MAP doesn't steal all packets!
 510         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 511               IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
 512               UDP(sport=20000, dport=10000) /
 513               Raw(b'\xa5' * 100))
 514         rx = self.send_and_expect(self.pg0, v4*1, self.pg0)
 515         v4_reply = v4[1]
 516         v4_reply.ttl -= 1
 517         for p in rx:
 518             self.validate(p[1], v4_reply)
 519         # Ensure MAP doesn't steal all packets
 520         v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 521               IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
 522               UDP(sport=20000, dport=10000) /
 523               Raw(b'\xa5' * 100))
 524         rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
 525         v6_reply = v6[1]
 526         v6_reply.hlim -= 1
 527         for p in rx:
 528             self.validate(p[1], v6_reply)
 529
 530         map_route = VppIpRoute(self,
 531                                "2001:db8::",
 532                                32,
 533                                [VppRoutePath(self.pg1.remote_ip6,
 534                                              self.pg1.sw_if_index,
 535                                              proto=DpoProto.DPO_PROTO_IP6)])
 536         map_route.add_vpp_config()
 537
 538         #
 539         # Send a v4 packet that will be translated
 540         #
 541         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 542         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 543         payload = TCP(sport=0xabcd, dport=0xabcd)
 544
 545         p4 = (p_ether / p_ip4 / payload)
 546         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 547                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 548         p6_translated.hlim -= 1
 549         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 550         for p in rx:
 551             self.validate(p[1], p6_translated)
 552
 553         # Send back an IPv6 packet that will be "untranslated"
 554         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 555         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 556                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 557         p6 = (p_ether6 / p_ip6 / payload)
 558         p4_translated = (IP(src='192.168.0.1',
 559                             dst=self.pg0.remote_ip4) / payload)
 560         p4_translated.id = 0
 561         p4_translated.ttl -= 1
 562         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 563         for p in rx:
 564             self.validate(p[1], p4_translated)
 565
 566         # IPv4 TTL=0
 567         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0)
 568         p4 = (p_ether / ip4_ttl_expired / payload)
 569
 570         icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
 571                           dst=self.pg0.remote_ip4) /
 572                        ICMP(type='time-exceeded',
 573                             code='ttl-zero-during-transit') /
 574                        IP(src=self.pg0.remote_ip4,
 575                           dst='192.168.0.1', ttl=0) / payload)
 576         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 577         for p in rx:
 578             self.validate(p[1], icmp4_reply)
 579
 580         # IPv4 TTL=1
 581         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=1)
 582         p4 = (p_ether / ip4_ttl_expired / payload)
 583
 584         icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
 585                           dst=self.pg0.remote_ip4) /
 586                        ICMP(type='time-exceeded',
 587                             code='ttl-zero-during-transit') /
 588                        IP(src=self.pg0.remote_ip4,
 589                           dst='192.168.0.1', ttl=1) / payload)
 590         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 591         for p in rx:
 592             self.validate(p[1], icmp4_reply)
 593
 594         # IPv6 Hop limit at BR
 595         ip6_hlim_expired = IPv6(hlim=1, src='2001:db8:1ab::c0a8:1:ab',
 596                                 dst='1234:5678:90ab:cdef:ac:1001:200:0')
 597         p6 = (p_ether6 / ip6_hlim_expired / payload)
 598
 599         icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
 600                             dst="2001:db8:1ab::c0a8:1:ab") /
 601                        ICMPv6TimeExceeded(code=0) /
 602                        IPv6(src="2001:db8:1ab::c0a8:1:ab",
 603                             dst='1234:5678:90ab:cdef:ac:1001:200:0',
 604                             hlim=1) / payload)
 605         rx = self.send_and_expect(self.pg1, p6*1, self.pg1)
 606         for p in rx:
 607             self.validate(p[1], icmp6_reply)
 608
 609         # IPv6 Hop limit beyond BR
 610         ip6_hlim_expired = IPv6(hlim=0, src='2001:db8:1ab::c0a8:1:ab',
 611                                 dst='1234:5678:90ab:cdef:ac:1001:200:0')
 612         p6 = (p_ether6 / ip6_hlim_expired / payload)
 613
 614         icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
 615                             dst="2001:db8:1ab::c0a8:1:ab") /
 616                        ICMPv6TimeExceeded(code=0) /
 617                        IPv6(src="2001:db8:1ab::c0a8:1:ab",
 618                             dst='1234:5678:90ab:cdef:ac:1001:200:0',
 619                             hlim=0) / payload)
 620         rx = self.send_and_expect(self.pg1, p6*1, self.pg1)
 621         for p in rx:
 622             self.validate(p[1], icmp6_reply)
 623
 624         # IPv4 Well-known port
 625         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 626         payload = UDP(sport=200, dport=200)
 627         p4 = (p_ether / p_ip4 / payload)
 628         self.send_and_assert_no_replies(self.pg0, p4*1)
 629
 630         # IPv6 Well-known port
 631         payload = UDP(sport=200, dport=200)
 632         p6 = (p_ether6 / p_ip6 / payload)
 633         self.send_and_assert_no_replies(self.pg1, p6*1)
 634
 635         # UDP packet fragmentation
 636         payload_len = 1453
 637         payload = UDP(sport=40000, dport=4000) / self.payload(payload_len)
 638         p4 = (p_ether / p_ip4 / payload)
 639         self.pg_enable_capture()
 640         self.pg0.add_stream(p4)
 641         self.pg_start()
 642         rx = self.pg1.get_capture(2)
 643
 644         p_ip6_translated = IPv6(src='1234:5678:90ab:cdef:ac:1001:200:0',
 645                                 dst='2001:db8:1e0::c0a8:1:e')
 646         for p in rx:
 647             self.validate_frag6(p, p_ip6_translated)
 648
 649         self.validate_frag_payload_len6(rx, UDP, payload_len)
 650
 651         # UDP packet fragmentation send fragments
 652         payload_len = 1453
 653         payload = UDP(sport=40000, dport=4000) / self.payload(payload_len)
 654         p4 = (p_ether / p_ip4 / payload)
 655         frags = fragment_rfc791(p4, fragsize=1000)
 656         self.pg_enable_capture()
 657         self.pg0.add_stream(frags)
 658         self.pg_start()
 659         rx = self.pg1.get_capture(2)
 660
 661         for p in rx:
 662             self.validate_frag6(p, p_ip6_translated)
 663
 664         self.validate_frag_payload_len6(rx, UDP, payload_len)
 665
 666         # Send back an fragmented IPv6 UDP packet that will be "untranslated"
 667         payload = UDP(sport=4000, dport=40000) / self.payload(payload_len)
 668         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 669         p_ip6 = IPv6(src='2001:db8:1e0::c0a8:1:e',
 670                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 671         p6 = (p_ether6 / p_ip6 / payload)
 672         frags6 = fragment_rfc8200(p6, identification=0xdcba, fragsize=1000)
 673
 674         p_ip4_translated = IP(src='192.168.0.1', dst=self.pg0.remote_ip4)
 675         p4_translated = (p_ip4_translated / payload)
 676         p4_translated.id = 0
 677         p4_translated.ttl -= 1
 678
 679         self.pg_enable_capture()
 680         self.pg1.add_stream(frags6)
 681         self.pg_start()
 682         rx = self.pg0.get_capture(2)
 683
 684         for p in rx:
 685             self.validate_frag4(p, p4_translated)
 686
 687         self.validate_frag_payload_len4(rx, UDP, payload_len)
 688
 689         # ICMP packet fragmentation
 690         payload = ICMP(id=6529) / self.payload(payload_len)
 691         p4 = (p_ether / p_ip4 / payload)
 692         self.pg_enable_capture()
 693         self.pg0.add_stream(p4)
 694         self.pg_start()
 695         rx = self.pg1.get_capture(2)
 696
 697         p_ip6_translated = IPv6(src='1234:5678:90ab:cdef:ac:1001:200:0',
 698                                 dst='2001:db8:160::c0a8:1:6')
 699         for p in rx:
 700             self.validate_frag6(p, p_ip6_translated)
 701
 702         self.validate_frag_payload_len6(rx, ICMPv6EchoRequest, payload_len)
 703
 704         # ICMP packet fragmentation send fragments
 705         payload = ICMP(id=6529) / self.payload(payload_len)
 706         p4 = (p_ether / p_ip4 / payload)
 707         frags = fragment_rfc791(p4, fragsize=1000)
 708         self.pg_enable_capture()
 709         self.pg0.add_stream(frags)
 710         self.pg_start()
 711         rx = self.pg1.get_capture(2)
 712
 713         for p in rx:
 714             self.validate_frag6(p, p_ip6_translated)
 715
 716         self.validate_frag_payload_len6(rx, ICMPv6EchoRequest, payload_len)
 717
 718         # TCP MSS clamping
 719         self.vapi.map_param_set_tcp(1300)
 720
 721         #
 722         # Send a v4 TCP SYN packet that will be translated and MSS clamped
 723         #
 724         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 725         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 726         payload = TCP(sport=0xabcd, dport=0xabcd, flags="S",
 727                       options=[('MSS', 1460)])
 728
 729         p4 = (p_ether / p_ip4 / payload)
 730         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 731                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 732         p6_translated.hlim -= 1
 733         p6_translated[TCP].options = [('MSS', 1300)]
 734         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 735         for p in rx:
 736             self.validate(p[1], p6_translated)
 737
 738         # Send back an IPv6 packet that will be "untranslated"
 739         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 740         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 741                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 742         p6 = (p_ether6 / p_ip6 / payload)
 743         p4_translated = (IP(src='192.168.0.1',
 744                             dst=self.pg0.remote_ip4) / payload)
 745         p4_translated.id = 0
 746         p4_translated.ttl -= 1
 747         p4_translated[TCP].options = [('MSS', 1300)]
 748         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 749         for p in rx:
 750             self.validate(p[1], p4_translated)
 751
 752         # TCP MSS clamping cleanup
 753         self.vapi.map_param_set_tcp(0)
 754
 755         # Enable icmp6 param to get back ICMPv6 unreachable messages in case
 756         # of security check fails
 757         self.vapi.map_param_set_icmp6(enable_unreachable=1)
 758
 759         # Send back an IPv6 packet that will be droppped due to security
 760         # check fail
 761         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 762         p_ip6_sec_check_fail = IPv6(src='2001:db8:1fe::c0a8:1:f',
 763                                     dst='1234:5678:90ab:cdef:ac:1001:200:0')
 764         payload = TCP(sport=0xabcd, dport=0xabcd)
 765         p6 = (p_ether6 / p_ip6_sec_check_fail / payload)
 766
 767         self.pg_send(self.pg1, p6*1)
 768         self.pg0.get_capture(0, timeout=1)
 769         rx = self.pg1.get_capture(1)
 770
 771         icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
 772                             dst='2001:db8:1fe::c0a8:1:f') /
 773                        ICMPv6DestUnreach(code=5) /
 774                        p_ip6_sec_check_fail / payload)
 775
 776         for p in rx:
 777             self.validate(p[1], icmp6_reply)
 778
 779         # ICMPv6 unreachable messages cleanup
 780         self.vapi.map_param_set_icmp6(enable_unreachable=0)
 781
 782     def test_map_t_ip6_psid(self):
 783         """ MAP-T v6->v4 PSID validation"""
 784
 785         #
 786         # Add a domain that maps from pg0 to pg1
 787         #
 788         map_dst = '2001:db8::/32'
 789         map_src = '1234:5678:90ab:cdef::/64'
 790         ip4_pfx = '192.168.0.0/24'
 791         tag = 'MAP-T Test Domain'
 792
 793         self.vapi.map_add_domain(ip6_prefix=map_dst,
 794                                  ip4_prefix=ip4_pfx,
 795                                  ip6_src=map_src,
 796                                  ea_bits_len=16,
 797                                  psid_offset=6,
 798                                  psid_length=4,
 799                                  mtu=1500,
 800                                  tag=tag)
 801
 802         # Enable MAP-T on interfaces.
 803         self.vapi.map_if_enable_disable(is_enable=1,
 804                                         sw_if_index=self.pg0.sw_if_index,
 805                                         is_translation=1)
 806         self.vapi.map_if_enable_disable(is_enable=1,
 807                                         sw_if_index=self.pg1.sw_if_index,
 808                                         is_translation=1)
 809
 810         map_route = VppIpRoute(self,
 811                                "2001:db8::",
 812                                32,
 813                                [VppRoutePath(self.pg1.remote_ip6,
 814                                              self.pg1.sw_if_index,
 815                                              proto=DpoProto.DPO_PROTO_IP6)])
 816         map_route.add_vpp_config()
 817
 818         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 819         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 820                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 821
 822         # Send good IPv6 source port, ensure translated IPv4 received
 823         payload = TCP(sport=0xabcd, dport=80)
 824         p6 = (p_ether6 / p_ip6 / payload)
 825         p4_translated = (IP(src='192.168.0.1',
 826                             dst=self.pg0.remote_ip4) / payload)
 827         p4_translated.id = 0
 828         p4_translated.ttl -= 1
 829         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 830         for p in rx:
 831             self.validate(p[1], p4_translated)
 832
 833         # Send bad IPv6 source port, ensure translated IPv4 not received
 834         payload = TCP(sport=0xdcba, dport=80)
 835         p6 = (p_ether6 / p_ip6 / payload)
 836         self.send_and_assert_no_replies(self.pg1, p6*1)
 837
 838     def test_map_t_pre_resolve(self):
 839         """ MAP-T pre-resolve"""
 840
 841         # Add a domain that maps from pg0 to pg1
 842         map_dst = '2001:db8::/32'
 843         map_src = '1234:5678:90ab:cdef::/64'
 844         ip4_pfx = '192.168.0.0/24'
 845         tag = 'MAP-T Test Domain.'
 846
 847         self.vapi.map_add_domain(ip6_prefix=map_dst,
 848                                  ip4_prefix=ip4_pfx,
 849                                  ip6_src=map_src,
 850                                  ea_bits_len=16,
 851                                  psid_offset=6,
 852                                  psid_length=4,
 853                                  mtu=1500,
 854                                  tag=tag)
 855
 856         # Enable MAP-T on interfaces.
 857         self.vapi.map_if_enable_disable(is_enable=1,
 858                                         sw_if_index=self.pg0.sw_if_index,
 859                                         is_translation=1)
 860         self.vapi.map_if_enable_disable(is_enable=1,
 861                                         sw_if_index=self.pg1.sw_if_index,
 862                                         is_translation=1)
 863
 864         # Enable pre-resolve option
 865         self.vapi.map_param_add_del_pre_resolve(ip4_nh_address="10.1.2.3",
 866                                                 ip6_nh_address="4001::1",
 867                                                 is_add=1)
 868
 869         # Add a route to 4001::1 and expect the translated traffic to be
 870         # sent via that route next-hop.
 871         pre_res_route6 = VppIpRoute(self, "4001::1", 128,
 872                                     [VppRoutePath(self.pg1.remote_hosts[2].ip6,
 873                                                   self.pg1.sw_if_index)])
 874         pre_res_route6.add_vpp_config()
 875
 876         # Add a route to 10.1.2.3 and expect the "untranslated" traffic to be
 877         # sent via that route next-hop.
 878         pre_res_route4 = VppIpRoute(self, "10.1.2.3", 32,
 879                                     [VppRoutePath(self.pg0.remote_hosts[1].ip4,
 880                                                   self.pg0.sw_if_index)])
 881         pre_res_route4.add_vpp_config()
 882
 883         # Send an IPv4 packet that will be translated
 884         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 885         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 886         payload = TCP(sport=0xabcd, dport=0xabcd)
 887         p4 = (p_ether / p_ip4 / payload)
 888
 889         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 890                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 891         p6_translated.hlim -= 1
 892
 893         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 894         for p in rx:
 895             self.assertEqual(p[Ether].dst, self.pg1.remote_hosts[2].mac)
 896             self.validate(p[1], p6_translated)
 897
 898         # Send back an IPv6 packet that will be "untranslated"
 899         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 900         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 901                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 902         p6 = (p_ether6 / p_ip6 / payload)
 903
 904         p4_translated = (IP(src='192.168.0.1',
 905                             dst=self.pg0.remote_ip4) / payload)
 906         p4_translated.id = 0
 907         p4_translated.ttl -= 1
 908
 909         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 910         for p in rx:
 911             self.assertEqual(p[Ether].dst, self.pg0.remote_hosts[1].mac)
 912             self.validate(p[1], p4_translated)
 913
 914         # Cleanup pre-resolve option
 915         self.vapi.map_param_add_del_pre_resolve(ip4_nh_address="10.1.2.3",
 916                                                 ip6_nh_address="4001::1",
 917                                                 is_add=0)
 918
 919 if __name__ == '__main__':
 920     unittest.main(testRunner=VppTestRunner)