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