test/test_map.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import socket
   5
   6 from framework import VppTestCase, VppTestRunner
   7 from vpp_ip import *
   8 from vpp_ip_route import VppIpRoute, VppRoutePath
   9
  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 = VppIp6Prefix(map_br_pfx, map_br_pfx_len).encode()
  80         map_src = VppIp6Prefix("3000::1", 128).encode()
  81         client_pfx = VppIp4Prefix("192.168.0.0", 16).encode()
  82         self.vapi.map_add_domain(map_dst, map_src, client_pfx)
  83
  84         #
  85         # Fire in a v4 packet that will be encapped to the BR
  86         #
  87         v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
  88               IP(src=self.pg0.remote_ip4, dst='192.168.1.1') /
  89               UDP(sport=20000, dport=10000) /
  90               Raw('\xa5' * 100))
  91
  92         self.send_and_assert_encapped(v4, "3000::1", "2001::c0a8:0:0")
  93
  94         #
  95         # Fire in a V6 encapped packet.
  96         #  expect a decapped packet on the inside ip4 link
  97         #
  98         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
  99              IPv6(dst='3000::1', src="2001::1") /
 100              IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
 101              UDP(sport=20000, dport=10000) /
 102              Raw('\xa5' * 100))
 103
 104         self.pg1.add_stream(p)
 105
 106         self.pg_enable_capture(self.pg_interfaces)
 107         self.pg_start()
 108
 109         rx = self.pg0.get_capture(1)
 110         rx = rx[0]
 111
 112         self.assertFalse(rx.haslayer(IPv6))
 113         self.assertEqual(rx[IP].src, p[IP].src)
 114         self.assertEqual(rx[IP].dst, p[IP].dst)
 115
 116         #
 117         # Pre-resolve. No API for this!!
 118         #
 119         self.vapi.ppcli("map params pre-resolve ip6-nh 4001::1")
 120
 121         self.send_and_assert_no_replies(self.pg0, v4,
 122                                         "resovled via default route")
 123
 124         #
 125         # Add a route to 4001::1. Expect the encapped traffic to be
 126         # sent via that routes next-hop
 127         #
 128         pre_res_route = VppIpRoute(
 129             self, "4001::1", 128,
 130             [VppRoutePath(self.pg1.remote_hosts[2].ip6,
 131                           self.pg1.sw_if_index,
 132                           proto=DpoProto.DPO_PROTO_IP6)],
 133             is_ip6=1)
 134         pre_res_route.add_vpp_config()
 135
 136         self.send_and_assert_encapped(v4, "3000::1",
 137                                       "2001::c0a8:0:0",
 138                                       dmac=self.pg1.remote_hosts[2].mac)
 139
 140         #
 141         # change the route to the pre-solved next-hop
 142         #
 143         pre_res_route.modify([VppRoutePath(self.pg1.remote_hosts[3].ip6,
 144                                            self.pg1.sw_if_index,
 145                                            proto=DpoProto.DPO_PROTO_IP6)])
 146         pre_res_route.add_vpp_config()
 147
 148         self.send_and_assert_encapped(v4, "3000::1",
 149                                       "2001::c0a8:0:0",
 150                                       dmac=self.pg1.remote_hosts[3].mac)
 151
 152         #
 153         # cleanup. The test infra's object registry will ensure
 154         # the route is really gone and thus that the unresolve worked.
 155         #
 156         pre_res_route.remove_vpp_config()
 157         self.vapi.ppcli("map params pre-resolve del ip6-nh 4001::1")
 158
 159     def validate(self, rx, expected):
 160         self.assertEqual(rx, expected.__class__(str(expected)))
 161
 162     def payload(self, len):
 163         return 'x' * len
 164
 165     def test_map_t(self):
 166         """ MAP-T """
 167
 168         #
 169         # Add a domain that maps from pg0 to pg1
 170         #
 171         map_dst = VppIp6Prefix("2001:db8::", 32).encode()
 172         map_src = VppIp6Prefix("1234:5678:90ab:cdef::", 64).encode()
 173         ip4_pfx = VppIp4Prefix("192.168.0.0", 24).encode()
 174         self.vapi.map_add_domain(map_dst, map_src, ip4_pfx, 16, 6, 4, 1)
 175
 176         # Enable MAP-T on interfaces.
 177
 178         # self.vapi.map_if_enable_disable(1, self.pg0.sw_if_index, 1)
 179         # self.vapi.map_if_enable_disable(1, self.pg1.sw_if_index, 1)
 180
 181         map_route = VppIpRoute(self,
 182                                "2001:db8::",
 183                                32,
 184                                [VppRoutePath(self.pg1.remote_ip6,
 185                                              self.pg1.sw_if_index,
 186                                              proto=DpoProto.DPO_PROTO_IP6)],
 187                                is_ip6=1)
 188         map_route.add_vpp_config()
 189
 190         #
 191         # Send a v4 packet that will be translated
 192         #
 193         p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
 194         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 195         payload = TCP(sport=0xabcd, dport=0xabcd)
 196
 197         p4 = (p_ether / p_ip4 / payload)
 198         p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
 199                               dst="2001:db8:1f0::c0a8:1:f") / payload)
 200         p6_translated.hlim -= 1
 201         rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
 202         for p in rx:
 203             self.validate(p[1], p6_translated)
 204
 205         # Send back an IPv6 packet that will be "untranslated"
 206         p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
 207         p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
 208                      dst='1234:5678:90ab:cdef:ac:1001:200:0')
 209         p6 = (p_ether6 / p_ip6 / payload)
 210         p4_translated = (IP(src='192.168.0.1',
 211                             dst=self.pg0.remote_ip4) / payload)
 212         p4_translated.id = 0
 213         p4_translated.ttl -= 1
 214         rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
 215         for p in rx:
 216             self.validate(p[1], p4_translated)
 217
 218         # IPv4 TTL
 219         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0)
 220         p4 = (p_ether / ip4_ttl_expired / payload)
 221
 222         icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
 223                           dst=self.pg0.remote_ip4) /
 224                        ICMP(type='time-exceeded',
 225                             code='ttl-zero-during-transit') /
 226                        IP(src=self.pg0.remote_ip4,
 227                           dst='192.168.0.1', ttl=0) / payload)
 228         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 229         for p in rx:
 230             self.validate(p[1], icmp4_reply)
 231
 232         '''
 233         This one is broken, cause it would require hairpinning...
 234         # IPv4 TTL TTL1
 235         ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=1)
 236         p4 = (p_ether / ip4_ttl_expired / payload)
 237
 238         icmp4_reply = IP(id=0, ttl=254, src=self.pg0.local_ip4,
 239         dst=self.pg0.remote_ip4) / \
 240         ICMP(type='time-exceeded', code='ttl-zero-during-transit' ) / \
 241         IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0) / payload
 242         rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
 243         for p in rx:
 244             self.validate(p[1], icmp4_reply)
 245         '''
 246
 247         # IPv6 Hop limit
 248         ip6_hlim_expired = IPv6(hlim=0, src='2001:db8:1ab::c0a8:1:ab',
 249                                 dst='1234:5678:90ab:cdef:ac:1001:200:0')
 250         p6 = (p_ether6 / ip6_hlim_expired / payload)
 251
 252         icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
 253                             dst="2001:db8:1ab::c0a8:1:ab") /
 254                        ICMPv6TimeExceeded(code=0) /
 255                        IPv6(src="2001:db8:1ab::c0a8:1:ab",
 256                             dst='1234:5678:90ab:cdef:ac:1001:200:0',
 257                             hlim=0) / payload)
 258         rx = self.send_and_expect(self.pg1, p6*1, self.pg1)
 259         for p in rx:
 260             self.validate(p[1], icmp6_reply)
 261
 262         # IPv4 Well-known port
 263         p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
 264         payload = UDP(sport=200, dport=200)
 265         p4 = (p_ether / p_ip4 / payload)
 266         self.send_and_assert_no_replies(self.pg0, p4*1)
 267
 268         # IPv6 Well-known port
 269         payload = UDP(sport=200, dport=200)
 270         p6 = (p_ether6 / p_ip6 / payload)
 271         self.send_and_assert_no_replies(self.pg1, p6*1)
 272
 273         # Packet fragmentation
 274         payload = UDP(sport=40000, dport=4000) / self.payload(1453)
 275         p4 = (p_ether / p_ip4 / payload)
 276         self.pg_enable_capture()
 277         self.pg0.add_stream(p4)
 278         self.pg_start()
 279         rx = self.pg1.get_capture(2)
 280         for p in rx:
 281             pass
 282             # TODO: Manual validation
 283             # self.validate(p[1], icmp4_reply)
 284
 285         # Packet fragmentation send fragments
 286         payload = UDP(sport=40000, dport=4000) / self.payload(1453)
 287         p4 = (p_ether / p_ip4 / payload)
 288         frags = fragment(p4, fragsize=1000)
 289         self.pg_enable_capture()
 290         self.pg0.add_stream(frags)
 291         self.pg_start()
 292         rx = self.pg1.get_capture(2)
 293         for p in rx:
 294             pass
 295             # p.show2()
 296         # reass_pkt = reassemble(rx)
 297         # p4_reply.ttl -= 1
 298         # p4_reply.id = 256
 299         # self.validate(reass_pkt, p4_reply)
 300
 301
 302 if __name__ == '__main__':
 303     unittest.main(testRunner=VppTestRunner)