test/test_map.py

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