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
  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         #
 236         # Pre-resolve. No API for this!!
 237         #
 238         self.vapi.ppcli("map params pre-resolve ip6-nh 4001::1")
 239
 240         self.send_and_assert_no_replies(self.pg0, v4,
 241                                         "resolved via default route")
 242
 243         #
 244         # Add a route to 4001::1. Expect the encapped traffic to be
 245         # sent via that routes next-hop
 246         #
 247         pre_res_route = VppIpRoute(self, "4001::1", 128,
 248                                    [VppRoutePath(self.pg1.remote_hosts[2].ip6,
 249                                                  self.pg1.sw_if_index)])
 250         pre_res_route.add_vpp_config()
 251
 252         self.send_and_assert_encapped_one(v4, "3000::1",
 253                                           map_translated_addr,
 254                                           dmac=self.pg1.remote_hosts[2].mac)
 255
 256         #
 257         # change the route to the pre-solved next-hop
 258         #
 259         pre_res_route.modify([VppRoutePath(self.pg1.remote_hosts[3].ip6,
 260                                            self.pg1.sw_if_index)])
 261         pre_res_route.add_vpp_config()
 262
 263         self.send_and_assert_encapped_one(v4, "3000::1",
 264                                           map_translated_addr,
 265                                           dmac=self.pg1.remote_hosts[3].mac)
 266
 267         #
 268         # cleanup. The test infra's object registry will ensure
 269         # the route is really gone and thus that the unresolve worked.
 270         #
 271         pre_res_route.remove_vpp_config()
 272         self.vapi.ppcli("map params pre-resolve del ip6-nh 4001::1")
 273
 274     def validate(self, rx, expected):
 275         self.assertEqual(rx, expected.__class__(scapy.compat.raw(expected)))
 276
 277     def payload(self, len):
 278         return 'x' * len
 279
 280     def test_map_t(self):
 281         """ MAP-T """
 282
 283         #
 284         # Add a domain that maps from pg0 to pg1
 285         #
 286         map_dst = '2001:db8::/32'
 287         map_src = '1234:5678:90ab:cdef::/64'
 288         ip4_pfx = '192.168.0.0/24'
 289         tag = 'MAP-T Tag.'
 290
 291         self.vapi.map_add_domain(ip6_prefix=map_dst,
 292                                  ip4_prefix=ip4_pfx,
 293                                  ip6_src=map_src,
 294                                  ea_bits_len=16,
 295                                  psid_offset=6,
 296                                  psid_length=4,
 297                                  mtu=1500,
 298                                  tag=tag)
 299
 300         # Enable MAP-T on interfaces.
 301         self.vapi.map_if_enable_disable(is_enable=1,
 302                                         sw_if_index=self.pg0.sw_if_index,
 303                                         is_translation=1)
 304         self.vapi.map_if_enable_disable(is_enable=1,
 305                                         sw_if_index=self.pg1.sw_if_index,
 306                                         is_translation=1)
 307
 308         # Ensure MAP doesn't steal all packets!
 309         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 310               IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
 311               UDP(sport=20000, dport=10000) /
 312               Raw('\xa5' * 100))
 313         rx = self.send_and_expect(self.pg0, v4*1, self.pg0)
 314         v4_reply = v4[1]
 315         v4_reply.ttl -= 1
 316         for p in rx:
 317             self.validate(p[1], v4_reply)
 318         # Ensure MAP doesn't steal all packets
 319         v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 320               IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
 321               UDP(sport=20000, dport=10000) /
 322               Raw('\xa5' * 100))
 323         rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
 324         v6_reply = v6[1]
 325         v6_reply.hlim -= 1
 326         for p in rx:
 327             self.validate(p[1], v6_reply)
 328
 329         map_route = VppIpRoute(self,
 330                                "2001:db8::",
 331                                32,
 332                                [VppRoutePath(self.pg1.remote_ip6,
 333                                              self.pg1.sw_if_index,
 334                                              proto=DpoProto.DPO_PROTO_IP6)])
 335         map_route.add_vpp_config()
 336
 337         #
 338         # Send a v4 packet that will be translated
 339         #
 340         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 341         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 342         payload = TCP(sport=0xabcd, dport=0xabcd)
 343
 344         p4 = (p_ether / p_ip4 / payload)
 345         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 346                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 347         p6_translated.hlim -= 1
 348         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 349         for p in rx:
 350             self.validate(p[1], p6_translated)
 351
 352         # Send back an IPv6 packet that will be "untranslated"
 353         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 354         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 355                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 356         p6 = (p_ether6 / p_ip6 / payload)
 357         p4_translated = (IP(src='192.168.0.1',
 358                             dst=self.pg0.remote_ip4) / payload)
 359         p4_translated.id = 0
 360         p4_translated.ttl -= 1
 361         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 362         for p in rx:
 363             self.validate(p[1], p4_translated)
 364
 365         # IPv4 TTL
 366         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0)
 367         p4 = (p_ether / ip4_ttl_expired / payload)
 368
 369         icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
 370                           dst=self.pg0.remote_ip4) /
 371                        ICMP(type='time-exceeded',
 372                             code='ttl-zero-during-transit') /
 373                        IP(src=self.pg0.remote_ip4,
 374                           dst='192.168.0.1', ttl=0) / payload)
 375         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 376         for p in rx:
 377             self.validate(p[1], icmp4_reply)
 378
 379         '''
 380         This one is broken, cause it would require hairpinning...
 381         # IPv4 TTL TTL1
 382         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=1)
 383         p4 = (p_ether / ip4_ttl_expired / payload)
 384
 385         icmp4_reply = IP(id=0, ttl=254, src=self.pg0.local_ip4,
 386         dst=self.pg0.remote_ip4) / \
 387         ICMP(type='time-exceeded', code='ttl-zero-during-transit' ) / \
 388         IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0) / payload
 389         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 390         for p in rx:
 391             self.validate(p[1], icmp4_reply)
 392         '''
 393
 394         # IPv6 Hop limit
 395         ip6_hlim_expired = IPv6(hlim=0, src='2001:db8:1ab::c0a8:1:ab',
 396                                 dst='1234:5678:90ab:cdef:ac:1001:200:0')
 397         p6 = (p_ether6 / ip6_hlim_expired / payload)
 398
 399         icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
 400                             dst="2001:db8:1ab::c0a8:1:ab") /
 401                        ICMPv6TimeExceeded(code=0) /
 402                        IPv6(src="2001:db8:1ab::c0a8:1:ab",
 403                             dst='1234:5678:90ab:cdef:ac:1001:200:0',
 404                             hlim=0) / payload)
 405         rx = self.send_and_expect(self.pg1, p6*1, self.pg1)
 406         for p in rx:
 407             self.validate(p[1], icmp6_reply)
 408
 409         # IPv4 Well-known port
 410         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 411         payload = UDP(sport=200, dport=200)
 412         p4 = (p_ether / p_ip4 / payload)
 413         self.send_and_assert_no_replies(self.pg0, p4*1)
 414
 415         # IPv6 Well-known port
 416         payload = UDP(sport=200, dport=200)
 417         p6 = (p_ether6 / p_ip6 / payload)
 418         self.send_and_assert_no_replies(self.pg1, p6*1)
 419
 420         # Packet fragmentation
 421         payload = UDP(sport=40000, dport=4000) / self.payload(1453)
 422         p4 = (p_ether / p_ip4 / payload)
 423         self.pg_enable_capture()
 424         self.pg0.add_stream(p4)
 425         self.pg_start()
 426         rx = self.pg1.get_capture(2)
 427         for p in rx:
 428             pass
 429             # TODO: Manual validation
 430             # self.validate(p[1], icmp4_reply)
 431
 432         # Packet fragmentation send fragments
 433         payload = UDP(sport=40000, dport=4000) / self.payload(1453)
 434         p4 = (p_ether / p_ip4 / payload)
 435         frags = fragment(p4, fragsize=1000)
 436         self.pg_enable_capture()
 437         self.pg0.add_stream(frags)
 438         self.pg_start()
 439         rx = self.pg1.get_capture(2)
 440         for p in rx:
 441             pass
 442             # p.show2()
 443
 444         # reass_pkt = reassemble(rx)
 445         # p4_reply.ttl -= 1
 446         # p4_reply.id = 256
 447         # self.validate(reass_pkt, p4_reply)
 448
 449         # TCP MSS clamping
 450         self.vapi.map_param_set_tcp(1300)
 451
 452         #
 453         # Send a v4 TCP SYN packet that will be translated and MSS clamped
 454         #
 455         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 456         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 457         payload = TCP(sport=0xabcd, dport=0xabcd, flags="S",
 458                       options=[('MSS', 1460)])
 459
 460         p4 = (p_ether / p_ip4 / payload)
 461         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 462                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 463         p6_translated.hlim -= 1
 464         p6_translated[TCP].options = [('MSS', 1300)]
 465         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 466         for p in rx:
 467             self.validate(p[1], p6_translated)
 468
 469         # Send back an IPv6 packet that will be "untranslated"
 470         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 471         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 472                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 473         p6 = (p_ether6 / p_ip6 / payload)
 474         p4_translated = (IP(src='192.168.0.1',
 475                             dst=self.pg0.remote_ip4) / payload)
 476         p4_translated.id = 0
 477         p4_translated.ttl -= 1
 478         p4_translated[TCP].options = [('MSS', 1300)]
 479         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 480         for p in rx:
 481             self.validate(p[1], p4_translated)
 482
 483
 484 if __name__ == '__main__':
 485     unittest.main(testRunner=VppTestRunner)