src/plugins/map/test/test_map.py

   1 #!/usr/bin/env python
   2
   3 import ipaddress
   4 import unittest
   5 from ipaddress import IPv6Network, IPv4Network
   6
   7 from framework import VppTestCase, VppTestRunner
   8 from vpp_ip import DpoProto
   9 from vpp_ip_route import VppIpRoute, VppRoutePath
  10 from util import fragment_rfc791, fragment_rfc8200
  11
  12 import scapy.compat
  13 from scapy.layers.l2 import Ether, Raw
  14 from scapy.layers.inet import IP, UDP, ICMP, TCP, fragment
  15 from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded
  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(self):
 101         """ MAP-E """
 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('\xa5' * 100))
 143         rx = self.send_and_expect(self.pg0, v4*1, 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('\xa5' * 100))
 156
 157         self.send_and_assert_encapped_one(v4, "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('\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('\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('\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('\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 validate(self, rx, expected):
 298         self.assertEqual(rx, expected.__class__(scapy.compat.raw(expected)))
 299
 300     def payload(self, len):
 301         return 'x' * len
 302
 303     def test_map_t(self):
 304         """ MAP-T """
 305
 306         #
 307         # Add a domain that maps from pg0 to pg1
 308         #
 309         map_dst = '2001:db8::/32'
 310         map_src = '1234:5678:90ab:cdef::/64'
 311         ip4_pfx = '192.168.0.0/24'
 312         tag = 'MAP-T Tag.'
 313
 314         self.vapi.map_add_domain(ip6_prefix=map_dst,
 315                                  ip4_prefix=ip4_pfx,
 316                                  ip6_src=map_src,
 317                                  ea_bits_len=16,
 318                                  psid_offset=6,
 319                                  psid_length=4,
 320                                  mtu=1500,
 321                                  tag=tag)
 322
 323         # Enable MAP-T on interfaces.
 324         self.vapi.map_if_enable_disable(is_enable=1,
 325                                         sw_if_index=self.pg0.sw_if_index,
 326                                         is_translation=1)
 327         self.vapi.map_if_enable_disable(is_enable=1,
 328                                         sw_if_index=self.pg1.sw_if_index,
 329                                         is_translation=1)
 330
 331         # Ensure MAP doesn't steal all packets!
 332         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 333               IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
 334               UDP(sport=20000, dport=10000) /
 335               Raw('\xa5' * 100))
 336         rx = self.send_and_expect(self.pg0, v4*1, self.pg0)
 337         v4_reply = v4[1]
 338         v4_reply.ttl -= 1
 339         for p in rx:
 340             self.validate(p[1], v4_reply)
 341         # Ensure MAP doesn't steal all packets
 342         v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 343               IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
 344               UDP(sport=20000, dport=10000) /
 345               Raw('\xa5' * 100))
 346         rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
 347         v6_reply = v6[1]
 348         v6_reply.hlim -= 1
 349         for p in rx:
 350             self.validate(p[1], v6_reply)
 351
 352         map_route = VppIpRoute(self,
 353                                "2001:db8::",
 354                                32,
 355                                [VppRoutePath(self.pg1.remote_ip6,
 356                                              self.pg1.sw_if_index,
 357                                              proto=DpoProto.DPO_PROTO_IP6)])
 358         map_route.add_vpp_config()
 359
 360         #
 361         # Send a v4 packet that will be translated
 362         #
 363         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 364         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 365         payload = TCP(sport=0xabcd, dport=0xabcd)
 366
 367         p4 = (p_ether / p_ip4 / payload)
 368         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 369                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 370         p6_translated.hlim -= 1
 371         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 372         for p in rx:
 373             self.validate(p[1], p6_translated)
 374
 375         # Send back an IPv6 packet that will be "untranslated"
 376         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 377         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 378                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 379         p6 = (p_ether6 / p_ip6 / payload)
 380         p4_translated = (IP(src='192.168.0.1',
 381                             dst=self.pg0.remote_ip4) / payload)
 382         p4_translated.id = 0
 383         p4_translated.ttl -= 1
 384         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 385         for p in rx:
 386             self.validate(p[1], p4_translated)
 387
 388         # IPv4 TTL
 389         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0)
 390         p4 = (p_ether / ip4_ttl_expired / payload)
 391
 392         icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
 393                           dst=self.pg0.remote_ip4) /
 394                        ICMP(type='time-exceeded',
 395                             code='ttl-zero-during-transit') /
 396                        IP(src=self.pg0.remote_ip4,
 397                           dst='192.168.0.1', ttl=0) / payload)
 398         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 399         for p in rx:
 400             self.validate(p[1], icmp4_reply)
 401
 402         '''
 403         This one is broken, cause it would require hairpinning...
 404         # IPv4 TTL TTL1
 405         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=1)
 406         p4 = (p_ether / ip4_ttl_expired / payload)
 407
 408         icmp4_reply = IP(id=0, ttl=254, src=self.pg0.local_ip4,
 409         dst=self.pg0.remote_ip4) / \
 410         ICMP(type='time-exceeded', code='ttl-zero-during-transit' ) / \
 411         IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0) / payload
 412         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 413         for p in rx:
 414             self.validate(p[1], icmp4_reply)
 415         '''
 416
 417         # IPv6 Hop limit
 418         ip6_hlim_expired = IPv6(hlim=0, src='2001:db8:1ab::c0a8:1:ab',
 419                                 dst='1234:5678:90ab:cdef:ac:1001:200:0')
 420         p6 = (p_ether6 / ip6_hlim_expired / payload)
 421
 422         icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
 423                             dst="2001:db8:1ab::c0a8:1:ab") /
 424                        ICMPv6TimeExceeded(code=0) /
 425                        IPv6(src="2001:db8:1ab::c0a8:1:ab",
 426                             dst='1234:5678:90ab:cdef:ac:1001:200:0',
 427                             hlim=0) / payload)
 428         rx = self.send_and_expect(self.pg1, p6*1, self.pg1)
 429         for p in rx:
 430             self.validate(p[1], icmp6_reply)
 431
 432         # IPv4 Well-known port
 433         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 434         payload = UDP(sport=200, dport=200)
 435         p4 = (p_ether / p_ip4 / payload)
 436         self.send_and_assert_no_replies(self.pg0, p4*1)
 437
 438         # IPv6 Well-known port
 439         payload = UDP(sport=200, dport=200)
 440         p6 = (p_ether6 / p_ip6 / payload)
 441         self.send_and_assert_no_replies(self.pg1, p6*1)
 442
 443         # Packet fragmentation
 444         payload = UDP(sport=40000, dport=4000) / self.payload(1453)
 445         p4 = (p_ether / p_ip4 / payload)
 446         self.pg_enable_capture()
 447         self.pg0.add_stream(p4)
 448         self.pg_start()
 449         rx = self.pg1.get_capture(2)
 450         for p in rx:
 451             pass
 452             # TODO: Manual validation
 453             # self.validate(p[1], icmp4_reply)
 454
 455         # Packet fragmentation send fragments
 456         payload = UDP(sport=40000, dport=4000) / self.payload(1453)
 457         p4 = (p_ether / p_ip4 / payload)
 458         frags = fragment(p4, fragsize=1000)
 459         self.pg_enable_capture()
 460         self.pg0.add_stream(frags)
 461         self.pg_start()
 462         rx = self.pg1.get_capture(2)
 463         for p in rx:
 464             pass
 465             # p.show2()
 466
 467         # reass_pkt = reassemble(rx)
 468         # p4_reply.ttl -= 1
 469         # p4_reply.id = 256
 470         # self.validate(reass_pkt, p4_reply)
 471
 472         # TCP MSS clamping
 473         self.vapi.map_param_set_tcp(1300)
 474
 475         #
 476         # Send a v4 TCP SYN packet that will be translated and MSS clamped
 477         #
 478         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 479         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 480         payload = TCP(sport=0xabcd, dport=0xabcd, flags="S",
 481                       options=[('MSS', 1460)])
 482
 483         p4 = (p_ether / p_ip4 / payload)
 484         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 485                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 486         p6_translated.hlim -= 1
 487         p6_translated[TCP].options = [('MSS', 1300)]
 488         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 489         for p in rx:
 490             self.validate(p[1], p6_translated)
 491
 492         # Send back an IPv6 packet that will be "untranslated"
 493         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 494         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 495                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 496         p6 = (p_ether6 / p_ip6 / payload)
 497         p4_translated = (IP(src='192.168.0.1',
 498                             dst=self.pg0.remote_ip4) / payload)
 499         p4_translated.id = 0
 500         p4_translated.ttl -= 1
 501         p4_translated[TCP].options = [('MSS', 1300)]
 502         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 503         for p in rx:
 504             self.validate(p[1], p4_translated)
 505
 506
 507 if __name__ == '__main__':
 508     unittest.main(testRunner=VppTestRunner)