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