test/test_map.py

   1 #!/usr/bin/env python
   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
  10 import scapy.compat
  11 from scapy.layers.l2 import Ether, Raw
  12 from scapy.layers.inet import IP, UDP, ICMP, TCP, fragment
  13 from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded
  14
  15
  16 class TestMAP(VppTestCase):
  17     """ MAP Test Case """
  18
  19     @classmethod
  20     def setUpClass(cls):
  21         super(TestMAP, cls).setUpClass()
  22
  23     @classmethod
  24     def tearDownClass(cls):
  25         super(TestMAP, cls).tearDownClass()
  26
  27     def setUp(self):
  28         super(TestMAP, self).setUp()
  29
  30         # create 2 pg interfaces
  31         self.create_pg_interfaces(range(4))
  32
  33         # pg0 is 'inside' IPv4
  34         self.pg0.admin_up()
  35         self.pg0.config_ip4()
  36         self.pg0.resolve_arp()
  37
  38         # pg1 is 'outside' IPv6
  39         self.pg1.admin_up()
  40         self.pg1.config_ip6()
  41         self.pg1.generate_remote_hosts(4)
  42         self.pg1.configure_ipv6_neighbors()
  43
  44     def tearDown(self):
  45         super(TestMAP, self).tearDown()
  46         for i in self.pg_interfaces:
  47             i.unconfig_ip4()
  48             i.unconfig_ip6()
  49             i.admin_down()
  50
  51     def send_and_assert_encapped(self, tx, ip6_src, ip6_dst, dmac=None):
  52         if not dmac:
  53             dmac = self.pg1.remote_mac
  54
  55         self.pg0.add_stream(tx)
  56
  57         self.pg_enable_capture(self.pg_interfaces)
  58         self.pg_start()
  59
  60         rx = self.pg1.get_capture(1)
  61         rx = rx[0]
  62
  63         self.assertEqual(rx[Ether].dst, dmac)
  64         self.assertEqual(rx[IP].src, tx[IP].src)
  65         self.assertEqual(rx[IPv6].src, ip6_src)
  66         self.assertEqual(rx[IPv6].dst, ip6_dst)
  67
  68     def test_api_map_domain_dump(self):
  69         map_dst = '2001::/64'
  70         map_src = '3000::1/128'
  71         client_pfx = '192.168.0.0/16'
  72         tag = 'MAP-E tag.'
  73         index = self.vapi.map_add_domain(ip4_prefix=client_pfx,
  74                                          ip6_prefix=map_dst,
  75                                          ip6_src=map_src,
  76                                          tag=tag).index
  77
  78         rv = self.vapi.map_domain_dump()
  79
  80         # restore the state early so as to not impact subsequent tests.
  81         # If an assert fails, we will not get the chance to do it at the end.
  82         self.vapi.map_del_domain(index=index)
  83
  84         self.assertGreater(len(rv), 0,
  85                            "Expected output from 'map_domain_dump'")
  86
  87         # typedefs are returned as ipaddress objects.
  88         # wrap results in str() ugh! to avoid the need to call unicode.
  89         self.assertEqual(str(rv[0].ip4_prefix), client_pfx)
  90         self.assertEqual(str(rv[0].ip6_prefix), map_dst)
  91         self.assertEqual(str(rv[0].ip6_src), map_src)
  92
  93         self.assertEqual(rv[0].tag, tag,
  94                          "output produced incorrect tag value.")
  95
  96     def test_map_e(self):
  97         """ MAP-E """
  98
  99         #
 100         # Add a route to the MAP-BR
 101         #
 102         map_br_pfx = "2001::"
 103         map_br_pfx_len = 64
 104         map_route = VppIpRoute(self,
 105                                map_br_pfx,
 106                                map_br_pfx_len,
 107                                [VppRoutePath(self.pg1.remote_ip6,
 108                                              self.pg1.sw_if_index,
 109                                              proto=DpoProto.DPO_PROTO_IP6)],
 110                                is_ip6=1)
 111         map_route.add_vpp_config()
 112
 113         #
 114         # Add a domain that maps from pg0 to pg1
 115         #
 116         map_dst = '2001::/64'
 117         map_src = '3000::1/128'
 118         client_pfx = '192.168.0.0/16'
 119         tag = 'MAP-E tag.'
 120         self.vapi.map_add_domain(ip4_prefix=client_pfx,
 121                                  ip6_prefix=map_dst,
 122                                  ip6_src=map_src,
 123                                  tag=tag)
 124
 125         # Enable MAP on interface.
 126         self.vapi.map_if_enable_disable(is_enable=1,
 127                                         sw_if_index=self.pg0.sw_if_index,
 128                                         is_translation=0)
 129
 130         # Ensure MAP doesn't steal all packets!
 131         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 132               IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
 133               UDP(sport=20000, dport=10000) /
 134               Raw('\xa5' * 100))
 135         rx = self.send_and_expect(self.pg0, v4*1, self.pg0)
 136         v4_reply = v4[1]
 137         v4_reply.ttl -= 1
 138         for p in rx:
 139             self.validate(p[1], v4_reply)
 140
 141         #
 142         # Fire in a v4 packet that will be encapped to the BR
 143         #
 144         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 145               IP(src=self.pg0.remote_ip4, dst='192.168.1.1') /
 146               UDP(sport=20000, dport=10000) /
 147               Raw('\xa5' * 100))
 148
 149         self.send_and_assert_encapped(v4, "3000::1", "2001::c0a8:0:0")
 150
 151         # Enable MAP on interface.
 152         self.vapi.map_if_enable_disable(is_enable=1,
 153                                         sw_if_index=self.pg1.sw_if_index,
 154                                         is_translation=0)
 155
 156         # Ensure MAP doesn't steal all packets
 157         v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 158               IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
 159               UDP(sport=20000, dport=10000) /
 160               Raw('\xa5' * 100))
 161         rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
 162         v6_reply = v6[1]
 163         v6_reply.hlim -= 1
 164         for p in rx:
 165             self.validate(p[1], v6_reply)
 166
 167         #
 168         # Fire in a V6 encapped packet.
 169         #  expect a decapped packet on the inside ip4 link
 170         #
 171         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 172              IPv6(dst='3000::1', src="2001::1") /
 173              IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
 174              UDP(sport=20000, dport=10000) /
 175              Raw('\xa5' * 100))
 176
 177         self.pg1.add_stream(p)
 178
 179         self.pg_enable_capture(self.pg_interfaces)
 180         self.pg_start()
 181
 182         rx = self.pg0.get_capture(1)
 183         rx = rx[0]
 184
 185         self.assertFalse(rx.haslayer(IPv6))
 186         self.assertEqual(rx[IP].src, p[IP].src)
 187         self.assertEqual(rx[IP].dst, p[IP].dst)
 188
 189         #
 190         # Pre-resolve. No API for this!!
 191         #
 192         self.vapi.ppcli("map params pre-resolve ip6-nh 4001::1")
 193
 194         self.send_and_assert_no_replies(self.pg0, v4,
 195                                         "resolved via default route")
 196
 197         #
 198         # Add a route to 4001::1. Expect the encapped traffic to be
 199         # sent via that routes next-hop
 200         #
 201         pre_res_route = VppIpRoute(
 202             self, "4001::1", 128,
 203             [VppRoutePath(self.pg1.remote_hosts[2].ip6,
 204                           self.pg1.sw_if_index,
 205                           proto=DpoProto.DPO_PROTO_IP6)],
 206             is_ip6=1)
 207         pre_res_route.add_vpp_config()
 208
 209         self.send_and_assert_encapped(v4, "3000::1",
 210                                       "2001::c0a8:0:0",
 211                                       dmac=self.pg1.remote_hosts[2].mac)
 212
 213         #
 214         # change the route to the pre-solved next-hop
 215         #
 216         pre_res_route.modify([VppRoutePath(self.pg1.remote_hosts[3].ip6,
 217                                            self.pg1.sw_if_index,
 218                                            proto=DpoProto.DPO_PROTO_IP6)])
 219         pre_res_route.add_vpp_config()
 220
 221         self.send_and_assert_encapped(v4, "3000::1",
 222                                       "2001::c0a8:0:0",
 223                                       dmac=self.pg1.remote_hosts[3].mac)
 224
 225         #
 226         # cleanup. The test infra's object registry will ensure
 227         # the route is really gone and thus that the unresolve worked.
 228         #
 229         pre_res_route.remove_vpp_config()
 230         self.vapi.ppcli("map params pre-resolve del ip6-nh 4001::1")
 231
 232     def validate(self, rx, expected):
 233         self.assertEqual(rx, expected.__class__(scapy.compat.raw(expected)))
 234
 235     def payload(self, len):
 236         return 'x' * len
 237
 238     def test_map_t(self):
 239         """ MAP-T """
 240
 241         #
 242         # Add a domain that maps from pg0 to pg1
 243         #
 244         map_dst = '2001:db8::/32'
 245         map_src = '1234:5678:90ab:cdef::/64'
 246         ip4_pfx = '192.168.0.0/24'
 247         tag = 'MAP-T Tag.'
 248
 249         self.vapi.map_add_domain(ip6_prefix=map_dst,
 250                                  ip4_prefix=ip4_pfx,
 251                                  ip6_src=map_src,
 252                                  ea_bits_len=16,
 253                                  psid_offset=6,
 254                                  psid_length=4,
 255                                  mtu=1500,
 256                                  tag=tag)
 257
 258         # Enable MAP-T on interfaces.
 259         self.vapi.map_if_enable_disable(is_enable=1,
 260                                         sw_if_index=self.pg0.sw_if_index,
 261                                         is_translation=1)
 262         self.vapi.map_if_enable_disable(is_enable=1,
 263                                         sw_if_index=self.pg1.sw_if_index,
 264                                         is_translation=1)
 265
 266         # Ensure MAP doesn't steal all packets!
 267         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 268               IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
 269               UDP(sport=20000, dport=10000) /
 270               Raw('\xa5' * 100))
 271         rx = self.send_and_expect(self.pg0, v4*1, self.pg0)
 272         v4_reply = v4[1]
 273         v4_reply.ttl -= 1
 274         for p in rx:
 275             self.validate(p[1], v4_reply)
 276         # Ensure MAP doesn't steal all packets
 277         v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 278               IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
 279               UDP(sport=20000, dport=10000) /
 280               Raw('\xa5' * 100))
 281         rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
 282         v6_reply = v6[1]
 283         v6_reply.hlim -= 1
 284         for p in rx:
 285             self.validate(p[1], v6_reply)
 286
 287         map_route = VppIpRoute(self,
 288                                "2001:db8::",
 289                                32,
 290                                [VppRoutePath(self.pg1.remote_ip6,
 291                                              self.pg1.sw_if_index,
 292                                              proto=DpoProto.DPO_PROTO_IP6)],
 293                                is_ip6=1)
 294         map_route.add_vpp_config()
 295
 296         #
 297         # Send a v4 packet that will be translated
 298         #
 299         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 300         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 301         payload = TCP(sport=0xabcd, dport=0xabcd)
 302
 303         p4 = (p_ether / p_ip4 / payload)
 304         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 305                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 306         p6_translated.hlim -= 1
 307         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 308         for p in rx:
 309             self.validate(p[1], p6_translated)
 310
 311         # Send back an IPv6 packet that will be "untranslated"
 312         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 313         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 314                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 315         p6 = (p_ether6 / p_ip6 / payload)
 316         p4_translated = (IP(src='192.168.0.1',
 317                             dst=self.pg0.remote_ip4) / payload)
 318         p4_translated.id = 0
 319         p4_translated.ttl -= 1
 320         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 321         for p in rx:
 322             self.validate(p[1], p4_translated)
 323
 324         # IPv4 TTL
 325         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0)
 326         p4 = (p_ether / ip4_ttl_expired / payload)
 327
 328         icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
 329                           dst=self.pg0.remote_ip4) /
 330                        ICMP(type='time-exceeded',
 331                             code='ttl-zero-during-transit') /
 332                        IP(src=self.pg0.remote_ip4,
 333                           dst='192.168.0.1', ttl=0) / payload)
 334         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 335         for p in rx:
 336             self.validate(p[1], icmp4_reply)
 337
 338         '''
 339         This one is broken, cause it would require hairpinning...
 340         # IPv4 TTL TTL1
 341         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=1)
 342         p4 = (p_ether / ip4_ttl_expired / payload)
 343
 344         icmp4_reply = IP(id=0, ttl=254, src=self.pg0.local_ip4,
 345         dst=self.pg0.remote_ip4) / \
 346         ICMP(type='time-exceeded', code='ttl-zero-during-transit' ) / \
 347         IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0) / payload
 348         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 349         for p in rx:
 350             self.validate(p[1], icmp4_reply)
 351         '''
 352
 353         # IPv6 Hop limit
 354         ip6_hlim_expired = IPv6(hlim=0, src='2001:db8:1ab::c0a8:1:ab',
 355                                 dst='1234:5678:90ab:cdef:ac:1001:200:0')
 356         p6 = (p_ether6 / ip6_hlim_expired / payload)
 357
 358         icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
 359                             dst="2001:db8:1ab::c0a8:1:ab") /
 360                        ICMPv6TimeExceeded(code=0) /
 361                        IPv6(src="2001:db8:1ab::c0a8:1:ab",
 362                             dst='1234:5678:90ab:cdef:ac:1001:200:0',
 363                             hlim=0) / payload)
 364         rx = self.send_and_expect(self.pg1, p6*1, self.pg1)
 365         for p in rx:
 366             self.validate(p[1], icmp6_reply)
 367
 368         # IPv4 Well-known port
 369         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 370         payload = UDP(sport=200, dport=200)
 371         p4 = (p_ether / p_ip4 / payload)
 372         self.send_and_assert_no_replies(self.pg0, p4*1)
 373
 374         # IPv6 Well-known port
 375         payload = UDP(sport=200, dport=200)
 376         p6 = (p_ether6 / p_ip6 / payload)
 377         self.send_and_assert_no_replies(self.pg1, p6*1)
 378
 379         # Packet fragmentation
 380         payload = UDP(sport=40000, dport=4000) / self.payload(1453)
 381         p4 = (p_ether / p_ip4 / payload)
 382         self.pg_enable_capture()
 383         self.pg0.add_stream(p4)
 384         self.pg_start()
 385         rx = self.pg1.get_capture(2)
 386         for p in rx:
 387             pass
 388             # TODO: Manual validation
 389             # self.validate(p[1], icmp4_reply)
 390
 391         # Packet fragmentation send fragments
 392         payload = UDP(sport=40000, dport=4000) / self.payload(1453)
 393         p4 = (p_ether / p_ip4 / payload)
 394         frags = fragment(p4, fragsize=1000)
 395         self.pg_enable_capture()
 396         self.pg0.add_stream(frags)
 397         self.pg_start()
 398         rx = self.pg1.get_capture(2)
 399         for p in rx:
 400             pass
 401             # p.show2()
 402         # reass_pkt = reassemble(rx)
 403         # p4_reply.ttl -= 1
 404         # p4_reply.id = 256
 405         # self.validate(reass_pkt, p4_reply)
 406
 407         # TCP MSS clamping
 408         self.vapi.map_param_set_tcp(1300)
 409
 410         #
 411         # Send a v4 TCP SYN packet that will be translated and MSS clamped
 412         #
 413         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 414         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 415         payload = TCP(sport=0xabcd, dport=0xabcd, flags="S",
 416                       options=[('MSS', 1460)])
 417
 418         p4 = (p_ether / p_ip4 / payload)
 419         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 420                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 421         p6_translated.hlim -= 1
 422         p6_translated[TCP].options = [('MSS', 1300)]
 423         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 424         for p in rx:
 425             self.validate(p[1], p6_translated)
 426
 427         # Send back an IPv6 packet that will be "untranslated"
 428         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 429         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 430                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 431         p6 = (p_ether6 / p_ip6 / payload)
 432         p4_translated = (IP(src='192.168.0.1',
 433                             dst=self.pg0.remote_ip4) / payload)
 434         p4_translated.id = 0
 435         p4_translated.ttl -= 1
 436         p4_translated[TCP].options = [('MSS', 1300)]
 437         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 438         for p in rx:
 439             self.validate(p[1], p4_translated)
 440
 441
 442 if __name__ == '__main__':
 443     unittest.main(testRunner=VppTestRunner)