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_frag(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_len(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 payload(self, len):
 462         return 'x' * len
 463
 464     def test_map_t(self):
 465         """ MAP-T """
 466
 467         #
 468         # Add a domain that maps from pg0 to pg1
 469         #
 470         map_dst = '2001:db8::/32'
 471         map_src = '1234:5678:90ab:cdef::/64'
 472         ip4_pfx = '192.168.0.0/24'
 473         tag = 'MAP-T Tag.'
 474
 475         self.vapi.map_add_domain(ip6_prefix=map_dst,
 476                                  ip4_prefix=ip4_pfx,
 477                                  ip6_src=map_src,
 478                                  ea_bits_len=16,
 479                                  psid_offset=6,
 480                                  psid_length=4,
 481                                  mtu=1500,
 482                                  tag=tag)
 483
 484         # Enable MAP-T on interfaces.
 485         self.vapi.map_if_enable_disable(is_enable=1,
 486                                         sw_if_index=self.pg0.sw_if_index,
 487                                         is_translation=1)
 488         self.vapi.map_if_enable_disable(is_enable=1,
 489                                         sw_if_index=self.pg1.sw_if_index,
 490                                         is_translation=1)
 491
 492         # Ensure MAP doesn't steal all packets!
 493         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 494               IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
 495               UDP(sport=20000, dport=10000) /
 496               Raw(b'\xa5' * 100))
 497         rx = self.send_and_expect(self.pg0, v4*1, self.pg0)
 498         v4_reply = v4[1]
 499         v4_reply.ttl -= 1
 500         for p in rx:
 501             self.validate(p[1], v4_reply)
 502         # Ensure MAP doesn't steal all packets
 503         v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 504               IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
 505               UDP(sport=20000, dport=10000) /
 506               Raw(b'\xa5' * 100))
 507         rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
 508         v6_reply = v6[1]
 509         v6_reply.hlim -= 1
 510         for p in rx:
 511             self.validate(p[1], v6_reply)
 512
 513         map_route = VppIpRoute(self,
 514                                "2001:db8::",
 515                                32,
 516                                [VppRoutePath(self.pg1.remote_ip6,
 517                                              self.pg1.sw_if_index,
 518                                              proto=DpoProto.DPO_PROTO_IP6)])
 519         map_route.add_vpp_config()
 520
 521         #
 522         # Send a v4 packet that will be translated
 523         #
 524         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 525         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 526         payload = TCP(sport=0xabcd, dport=0xabcd)
 527
 528         p4 = (p_ether / p_ip4 / payload)
 529         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 530                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 531         p6_translated.hlim -= 1
 532         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 533         for p in rx:
 534             self.validate(p[1], p6_translated)
 535
 536         # Send back an IPv6 packet that will be "untranslated"
 537         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 538         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 539                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 540         p6 = (p_ether6 / p_ip6 / payload)
 541         p4_translated = (IP(src='192.168.0.1',
 542                             dst=self.pg0.remote_ip4) / payload)
 543         p4_translated.id = 0
 544         p4_translated.ttl -= 1
 545         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 546         for p in rx:
 547             self.validate(p[1], p4_translated)
 548
 549         # IPv4 TTL=0
 550         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0)
 551         p4 = (p_ether / ip4_ttl_expired / payload)
 552
 553         icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
 554                           dst=self.pg0.remote_ip4) /
 555                        ICMP(type='time-exceeded',
 556                             code='ttl-zero-during-transit') /
 557                        IP(src=self.pg0.remote_ip4,
 558                           dst='192.168.0.1', ttl=0) / payload)
 559         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 560         for p in rx:
 561             self.validate(p[1], icmp4_reply)
 562
 563         # IPv4 TTL=1
 564         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=1)
 565         p4 = (p_ether / ip4_ttl_expired / payload)
 566
 567         icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
 568                           dst=self.pg0.remote_ip4) /
 569                        ICMP(type='time-exceeded',
 570                             code='ttl-zero-during-transit') /
 571                        IP(src=self.pg0.remote_ip4,
 572                           dst='192.168.0.1', ttl=1) / payload)
 573         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 574         for p in rx:
 575             self.validate(p[1], icmp4_reply)
 576
 577         # IPv6 Hop limit
 578         ip6_hlim_expired = IPv6(hlim=0, src='2001:db8:1ab::c0a8:1:ab',
 579                                 dst='1234:5678:90ab:cdef:ac:1001:200:0')
 580         p6 = (p_ether6 / ip6_hlim_expired / payload)
 581
 582         icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
 583                             dst="2001:db8:1ab::c0a8:1:ab") /
 584                        ICMPv6TimeExceeded(code=0) /
 585                        IPv6(src="2001:db8:1ab::c0a8:1:ab",
 586                             dst='1234:5678:90ab:cdef:ac:1001:200:0',
 587                             hlim=0) / payload)
 588         rx = self.send_and_expect(self.pg1, p6*1, self.pg1)
 589         for p in rx:
 590             self.validate(p[1], icmp6_reply)
 591
 592         # IPv4 Well-known port
 593         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 594         payload = UDP(sport=200, dport=200)
 595         p4 = (p_ether / p_ip4 / payload)
 596         self.send_and_assert_no_replies(self.pg0, p4*1)
 597
 598         # IPv6 Well-known port
 599         payload = UDP(sport=200, dport=200)
 600         p6 = (p_ether6 / p_ip6 / payload)
 601         self.send_and_assert_no_replies(self.pg1, p6*1)
 602
 603         # UDP packet fragmentation
 604         payload_len = 1453
 605         payload = UDP(sport=40000, dport=4000) / self.payload(payload_len)
 606         p4 = (p_ether / p_ip4 / payload)
 607         self.pg_enable_capture()
 608         self.pg0.add_stream(p4)
 609         self.pg_start()
 610         rx = self.pg1.get_capture(2)
 611
 612         p_ip6_translated = IPv6(src='1234:5678:90ab:cdef:ac:1001:200:0',
 613                                 dst='2001:db8:1e0::c0a8:1:e')
 614         for p in rx:
 615             self.validate_frag(p, p_ip6_translated)
 616
 617         self.validate_frag_payload_len(rx, UDP, payload_len)
 618
 619         # UDP packet fragmentation send fragments
 620         payload = UDP(sport=40000, dport=4000) / self.payload(payload_len)
 621         p4 = (p_ether / p_ip4 / payload)
 622         frags = fragment_rfc791(p4, fragsize=1000)
 623         self.pg_enable_capture()
 624         self.pg0.add_stream(frags)
 625         self.pg_start()
 626         rx = self.pg1.get_capture(2)
 627
 628         for p in rx:
 629             self.validate_frag(p, p_ip6_translated)
 630
 631         self.validate_frag_payload_len(rx, UDP, payload_len)
 632
 633         # ICMP packet fragmentation
 634         payload = ICMP(id=6529) / self.payload(payload_len)
 635         p4 = (p_ether / p_ip4 / payload)
 636         self.pg_enable_capture()
 637         self.pg0.add_stream(p4)
 638         self.pg_start()
 639         rx = self.pg1.get_capture(2)
 640
 641         p_ip6_translated = IPv6(src='1234:5678:90ab:cdef:ac:1001:200:0',
 642                                 dst='2001:db8:160::c0a8:1:6')
 643         for p in rx:
 644             self.validate_frag(p, p_ip6_translated)
 645
 646         self.validate_frag_payload_len(rx, ICMPv6EchoRequest, payload_len)
 647
 648         # ICMP packet fragmentation send fragments
 649         payload = ICMP(id=6529) / self.payload(payload_len)
 650         p4 = (p_ether / p_ip4 / payload)
 651         frags = fragment_rfc791(p4, fragsize=1000)
 652         self.pg_enable_capture()
 653         self.pg0.add_stream(frags)
 654         self.pg_start()
 655         rx = self.pg1.get_capture(2)
 656
 657         for p in rx:
 658             self.validate_frag(p, p_ip6_translated)
 659
 660         self.validate_frag_payload_len(rx, ICMPv6EchoRequest, payload_len)
 661
 662         # TCP MSS clamping
 663         self.vapi.map_param_set_tcp(1300)
 664
 665         #
 666         # Send a v4 TCP SYN packet that will be translated and MSS clamped
 667         #
 668         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 669         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 670         payload = TCP(sport=0xabcd, dport=0xabcd, flags="S",
 671                       options=[('MSS', 1460)])
 672
 673         p4 = (p_ether / p_ip4 / payload)
 674         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 675                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 676         p6_translated.hlim -= 1
 677         p6_translated[TCP].options = [('MSS', 1300)]
 678         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 679         for p in rx:
 680             self.validate(p[1], p6_translated)
 681
 682         # Send back an IPv6 packet that will be "untranslated"
 683         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 684         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 685                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 686         p6 = (p_ether6 / p_ip6 / payload)
 687         p4_translated = (IP(src='192.168.0.1',
 688                             dst=self.pg0.remote_ip4) / payload)
 689         p4_translated.id = 0
 690         p4_translated.ttl -= 1
 691         p4_translated[TCP].options = [('MSS', 1300)]
 692         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 693         for p in rx:
 694             self.validate(p[1], p4_translated)
 695
 696         # TCP MSS clamping cleanup
 697         self.vapi.map_param_set_tcp(0)
 698
 699         # Enable icmp6 param to get back ICMPv6 unreachable messages in case
 700         # of security check fails
 701         self.vapi.map_param_set_icmp6(enable_unreachable=1)
 702
 703         # Send back an IPv6 packet that will be droppped due to security
 704         # check fail
 705         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 706         p_ip6_sec_check_fail = IPv6(src='2001:db8:1fe::c0a8:1:f',
 707                                     dst='1234:5678:90ab:cdef:ac:1001:200:0')
 708         payload = TCP(sport=0xabcd, dport=0xabcd)
 709         p6 = (p_ether6 / p_ip6_sec_check_fail / payload)
 710
 711         self.pg_send(self.pg1, p6*1)
 712         self.pg0.get_capture(0, timeout=1)
 713         rx = self.pg1.get_capture(1)
 714
 715         icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
 716                             dst='2001:db8:1fe::c0a8:1:f') /
 717                        ICMPv6DestUnreach(code=5) /
 718                        p_ip6_sec_check_fail / payload)
 719
 720         for p in rx:
 721             self.validate(p[1], icmp6_reply)
 722
 723         # ICMPv6 unreachable messages cleanup
 724         self.vapi.map_param_set_icmp6(enable_unreachable=0)
 725
 726     def test_map_t_ip6_psid(self):
 727         """ MAP-T v6->v4 PSID validation"""
 728
 729         #
 730         # Add a domain that maps from pg0 to pg1
 731         #
 732         map_dst = '2001:db8::/32'
 733         map_src = '1234:5678:90ab:cdef::/64'
 734         ip4_pfx = '192.168.0.0/24'
 735         tag = 'MAP-T Test Domain'
 736
 737         self.vapi.map_add_domain(ip6_prefix=map_dst,
 738                                  ip4_prefix=ip4_pfx,
 739                                  ip6_src=map_src,
 740                                  ea_bits_len=16,
 741                                  psid_offset=6,
 742                                  psid_length=4,
 743                                  mtu=1500,
 744                                  tag=tag)
 745
 746         # Enable MAP-T on interfaces.
 747         self.vapi.map_if_enable_disable(is_enable=1,
 748                                         sw_if_index=self.pg0.sw_if_index,
 749                                         is_translation=1)
 750         self.vapi.map_if_enable_disable(is_enable=1,
 751                                         sw_if_index=self.pg1.sw_if_index,
 752                                         is_translation=1)
 753
 754         map_route = VppIpRoute(self,
 755                                "2001:db8::",
 756                                32,
 757                                [VppRoutePath(self.pg1.remote_ip6,
 758                                              self.pg1.sw_if_index,
 759                                              proto=DpoProto.DPO_PROTO_IP6)])
 760         map_route.add_vpp_config()
 761
 762         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 763         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 764                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 765
 766         # Send good IPv6 source port, ensure translated IPv4 received
 767         payload = TCP(sport=0xabcd, dport=80)
 768         p6 = (p_ether6 / p_ip6 / payload)
 769         p4_translated = (IP(src='192.168.0.1',
 770                             dst=self.pg0.remote_ip4) / payload)
 771         p4_translated.id = 0
 772         p4_translated.ttl -= 1
 773         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 774         for p in rx:
 775             self.validate(p[1], p4_translated)
 776
 777         # Send bad IPv6 source port, ensure translated IPv4 not received
 778         payload = TCP(sport=0xdcba, dport=80)
 779         p6 = (p_ether6 / p_ip6 / payload)
 780         self.send_and_assert_no_replies(self.pg1, p6*1)
 781
 782     def test_map_t_pre_resolve(self):
 783         """ MAP-T pre-resolve"""
 784
 785         # Add a domain that maps from pg0 to pg1
 786         map_dst = '2001:db8::/32'
 787         map_src = '1234:5678:90ab:cdef::/64'
 788         ip4_pfx = '192.168.0.0/24'
 789         tag = 'MAP-T Test Domain.'
 790
 791         self.vapi.map_add_domain(ip6_prefix=map_dst,
 792                                  ip4_prefix=ip4_pfx,
 793                                  ip6_src=map_src,
 794                                  ea_bits_len=16,
 795                                  psid_offset=6,
 796                                  psid_length=4,
 797                                  mtu=1500,
 798                                  tag=tag)
 799
 800         # Enable MAP-T on interfaces.
 801         self.vapi.map_if_enable_disable(is_enable=1,
 802                                         sw_if_index=self.pg0.sw_if_index,
 803                                         is_translation=1)
 804         self.vapi.map_if_enable_disable(is_enable=1,
 805                                         sw_if_index=self.pg1.sw_if_index,
 806                                         is_translation=1)
 807
 808         # Enable pre-resolve option
 809         self.vapi.map_param_add_del_pre_resolve(ip4_nh_address="10.1.2.3",
 810                                                 ip6_nh_address="4001::1",
 811                                                 is_add=1)
 812
 813         # Add a route to 4001::1 and expect the translated traffic to be
 814         # sent via that route next-hop.
 815         pre_res_route6 = VppIpRoute(self, "4001::1", 128,
 816                                     [VppRoutePath(self.pg1.remote_hosts[2].ip6,
 817                                                   self.pg1.sw_if_index)])
 818         pre_res_route6.add_vpp_config()
 819
 820         # Add a route to 10.1.2.3 and expect the "untranslated" traffic to be
 821         # sent via that route next-hop.
 822         pre_res_route4 = VppIpRoute(self, "10.1.2.3", 32,
 823                                     [VppRoutePath(self.pg0.remote_hosts[1].ip4,
 824                                                   self.pg0.sw_if_index)])
 825         pre_res_route4.add_vpp_config()
 826
 827         # Send an IPv4 packet that will be translated
 828         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 829         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 830         payload = TCP(sport=0xabcd, dport=0xabcd)
 831         p4 = (p_ether / p_ip4 / payload)
 832
 833         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 834                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 835         p6_translated.hlim -= 1
 836
 837         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 838         for p in rx:
 839             self.assertEqual(p[Ether].dst, self.pg1.remote_hosts[2].mac)
 840             self.validate(p[1], p6_translated)
 841
 842         # Send back an IPv6 packet that will be "untranslated"
 843         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 844         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 845                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 846         p6 = (p_ether6 / p_ip6 / payload)
 847
 848         p4_translated = (IP(src='192.168.0.1',
 849                             dst=self.pg0.remote_ip4) / payload)
 850         p4_translated.id = 0
 851         p4_translated.ttl -= 1
 852
 853         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 854         for p in rx:
 855             self.assertEqual(p[Ether].dst, self.pg0.remote_hosts[1].mac)
 856             self.validate(p[1], p4_translated)
 857
 858         # Cleanup pre-resolve option
 859         self.vapi.map_param_add_del_pre_resolve(ip4_nh_address="10.1.2.3",
 860                                                 ip6_nh_address="4001::1",
 861                                                 is_add=0)
 862
 863 if __name__ == '__main__':
 864     unittest.main(testRunner=VppTestRunner)