test/test_srv6_un.py

   1 #!/usr/bin/env python
   2
   3 from framework import VppTestCase
   4 from ipaddress import IPv4Address
   5 from ipaddress import IPv6Address
   6 from scapy.contrib.gtp import *
   7 from scapy.all import *
   8
   9
  10 class TestSRv6uSIDSRH(VppTestCase):
  11     """ SRv6 End.uSID w/ SRH """
  12
  13     @classmethod
  14     def setUpClass(cls):
  15         super(TestSRv6uSIDSRH, cls).setUpClass()
  16         try:
  17             cls.create_pg_interfaces(range(2))
  18             cls.pg_if_i = cls.pg_interfaces[0]
  19             cls.pg_if_o = cls.pg_interfaces[1]
  20
  21             cls.pg_if_i.config_ip6()
  22             cls.pg_if_o.config_ip6()
  23
  24             cls.ip6_nhop = cls.pg_if_o.remote_ip6
  25
  26             cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
  27             cls.ip6_src = "1111:2222::1"
  28
  29             for pg_if in cls.pg_interfaces:
  30                 pg_if.admin_up()
  31                 pg_if.resolve_ndp(timeout=5)
  32
  33         except Exception:
  34             super(TestSRv6uSIDSRH, cls).tearDownClass()
  35             raise
  36
  37     def create_packets(self, inner):
  38
  39         ip6_dst = IPv6Address(str(self.ip6_dst))
  40
  41         ip6_src = IPv6Address(str(self.ip6_src))
  42
  43         self.logger.info("ip6 dst: {}".format(ip6_dst))
  44         self.logger.info("ip6 src: {}".format(ip6_src))
  45
  46         pkts = list()
  47         for d, s in inner:
  48             pkt = (Ether() /
  49                    IPv6(dst=str(ip6_dst), src=str(ip6_src)) /
  50                    IPv6ExtHdrSegmentRouting(segleft=1,
  51                                             lastentry=0,
  52                                             tag=0,
  53                                             addresses=[
  54                                                 "a1::1",
  55                                                 "1111:2222:aaaa:bbbb::"]) /
  56                    IPv6(dst=d, src=s) /
  57                    UDP(sport=1000, dport=23))
  58             self.logger.info(pkt.show2(dump=True))
  59             pkts.append(pkt)
  60
  61         return pkts
  62
  63     def test_srv6_usid_srh(self):
  64         """ test_srv6_usid_srh """
  65         pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
  66
  67         self.vapi.cli("set sr encaps source addr A1::1")
  68         self.vapi.cli(
  69             "sr localsid prefix 1111:2222:aaaa::/48 behavior un 16")
  70         self.vapi.cli(
  71             "ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
  72
  73         self.logger.info(self.vapi.cli("show sr localsids"))
  74
  75         self.vapi.cli("clear errors")
  76
  77         self.pg0.add_stream(pkts)
  78         self.pg_enable_capture(self.pg_interfaces)
  79         self.pg_start()
  80
  81         self.logger.info(self.vapi.cli("show errors"))
  82         self.logger.info(self.vapi.cli("show int address"))
  83
  84         capture = self.pg1.get_capture(len(pkts))
  85
  86         for pkt in capture:
  87             self.logger.info(pkt.show2(dump=True))
  88             self.assertEqual(
  89                     pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
  90
  91
  92 class TestSRv6uSID(VppTestCase):
  93     """ SRv6 End.uSID w/o SRH """
  94
  95     @classmethod
  96     def setUpClass(cls):
  97         super(TestSRv6uSID, cls).setUpClass()
  98         try:
  99             cls.create_pg_interfaces(range(2))
 100             cls.pg_if_i = cls.pg_interfaces[0]
 101             cls.pg_if_o = cls.pg_interfaces[1]
 102
 103             cls.pg_if_i.config_ip6()
 104             cls.pg_if_o.config_ip6()
 105
 106             cls.ip6_nhop = cls.pg_if_o.remote_ip6
 107
 108             cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
 109             cls.ip6_src = "1111:2222::1"
 110
 111             for pg_if in cls.pg_interfaces:
 112                 pg_if.admin_up()
 113                 pg_if.resolve_ndp(timeout=5)
 114
 115         except Exception:
 116             super(TestSRv6uSID, cls).tearDownClass()
 117             raise
 118
 119     def create_packets(self, inner):
 120
 121         ip6_dst = IPv6Address(str(self.ip6_dst))
 122
 123         ip6_src = IPv6Address(str(self.ip6_src))
 124
 125         self.logger.info("ip6 dst: {}".format(ip6_dst))
 126         self.logger.info("ip6 src: {}".format(ip6_src))
 127
 128         pkts = list()
 129         for d, s in inner:
 130             pkt = (Ether() /
 131                    IPv6(dst=str(ip6_dst), src=str(ip6_src)) /
 132                    IPv6(dst=d, src=s) /
 133                    UDP(sport=1000, dport=23))
 134             self.logger.info(pkt.show2(dump=True))
 135             pkts.append(pkt)
 136
 137         return pkts
 138
 139     def test_srv6_usid(self):
 140         """ test_srv6_usid """
 141         pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
 142
 143         self.vapi.cli("set sr encaps source addr A1::1")
 144         self.vapi.cli(
 145             "sr localsid prefix 1111:2222:aaaa::/48 behavior un 16")
 146         self.vapi.cli(
 147             "ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
 148
 149         self.logger.info(self.vapi.cli("show sr localsids"))
 150
 151         self.vapi.cli("clear errors")
 152
 153         self.pg0.add_stream(pkts)
 154         self.pg_enable_capture(self.pg_interfaces)
 155         self.pg_start()
 156
 157         self.logger.info(self.vapi.cli("show errors"))
 158         self.logger.info(self.vapi.cli("show int address"))
 159
 160         capture = self.pg1.get_capture(len(pkts))
 161
 162         for pkt in capture:
 163             self.logger.info(pkt.show2(dump=True))
 164             self.assertEqual(
 165                     pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
 166
 167
 168 class TestSRv6uSIDFlexSRH(VppTestCase):
 169     """ SRv6 End.uSID.Flex w/ SRH """
 170
 171     @classmethod
 172     def setUpClass(cls):
 173         super(TestSRv6uSIDFlexSRH, cls).setUpClass()
 174         try:
 175             cls.create_pg_interfaces(range(2))
 176             cls.pg_if_i = cls.pg_interfaces[0]
 177             cls.pg_if_o = cls.pg_interfaces[1]
 178
 179             cls.pg_if_i.config_ip6()
 180             cls.pg_if_o.config_ip6()
 181
 182             cls.ip6_nhop = cls.pg_if_o.remote_ip6
 183
 184             cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
 185             cls.ip6_src = "1111:2222::1"
 186
 187             for pg_if in cls.pg_interfaces:
 188                 pg_if.admin_up()
 189                 pg_if.resolve_ndp(timeout=5)
 190
 191         except Exception:
 192             super(TestSRv6uSIDFlexSRH, cls).tearDownClass()
 193             raise
 194
 195     def create_packets(self, inner):
 196
 197         ip6_dst = IPv6Address(str(self.ip6_dst))
 198
 199         ip6_src = IPv6Address(str(self.ip6_src))
 200
 201         self.logger.info("ip6 dst: {}".format(ip6_dst))
 202         self.logger.info("ip6 src: {}".format(ip6_src))
 203
 204         pkts = list()
 205         for d, s in inner:
 206             pkt = (Ether() /
 207                    IPv6(dst=str(ip6_dst), src=str(ip6_src)) /
 208                    IPv6ExtHdrSegmentRouting(segleft=1,
 209                                             lastentry=0,
 210                                             tag=0,
 211                                             addresses=[
 212                                                 "a1::1",
 213                                                 "1111:2222:aaaa:bbbb::"]) /
 214                    IPv6(dst=d, src=s) /
 215                    UDP(sport=1000, dport=23))
 216             self.logger.info(pkt.show2(dump=True))
 217             pkts.append(pkt)
 218
 219         return pkts
 220
 221     def test_srv6_usid_flex_srh(self):
 222         """ test_srv6_usid_flex_srh """
 223         pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
 224
 225         self.vapi.cli("set sr encaps source addr A1::1")
 226         self.vapi.cli(
 227             "sr localsid prefix 1111:2222:aaaa::/48 behavior un.flex 16")
 228         self.vapi.cli(
 229             "ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
 230
 231         self.logger.info(self.vapi.cli("show sr localsids"))
 232
 233         self.vapi.cli("clear errors")
 234
 235         self.pg0.add_stream(pkts)
 236         self.pg_enable_capture(self.pg_interfaces)
 237         self.pg_start()
 238
 239         self.logger.info(self.vapi.cli("show errors"))
 240         self.logger.info(self.vapi.cli("show int address"))
 241
 242         capture = self.pg1.get_capture(len(pkts))
 243
 244         for pkt in capture:
 245             self.logger.info(pkt.show2(dump=True))
 246             self.assertEqual(
 247                     pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")
 248
 249
 250 class TestSRv6uSIDFlex(VppTestCase):
 251     """ SRv6 End.uSID.Flex w/o SRH """
 252
 253     @classmethod
 254     def setUpClass(cls):
 255         super(TestSRv6uSIDFlex, cls).setUpClass()
 256         try:
 257             cls.create_pg_interfaces(range(2))
 258             cls.pg_if_i = cls.pg_interfaces[0]
 259             cls.pg_if_o = cls.pg_interfaces[1]
 260
 261             cls.pg_if_i.config_ip6()
 262             cls.pg_if_o.config_ip6()
 263
 264             cls.ip6_nhop = cls.pg_if_o.remote_ip6
 265
 266             cls.ip6_dst = "1111:2222:aaaa:bbbb:cccc:dddd:eeee:ffff"
 267             cls.ip6_src = "1111:2222::1"
 268
 269             for pg_if in cls.pg_interfaces:
 270                 pg_if.admin_up()
 271                 pg_if.resolve_ndp(timeout=5)
 272
 273         except Exception:
 274             super(TestSRv6uSIDFlex, cls).tearDownClass()
 275             raise
 276
 277     def create_packets(self, inner):
 278
 279         ip6_dst = IPv6Address(str(self.ip6_dst))
 280
 281         ip6_src = IPv6Address(str(self.ip6_src))
 282
 283         self.logger.info("ip6 dst: {}".format(ip6_dst))
 284         self.logger.info("ip6 src: {}".format(ip6_src))
 285
 286         pkts = list()
 287         for d, s in inner:
 288             pkt = (Ether() /
 289                    IPv6(dst=str(ip6_dst), src=str(ip6_src)) /
 290                    IPv6(dst=d, src=s) /
 291                    UDP(sport=1000, dport=23))
 292             self.logger.info(pkt.show2(dump=True))
 293             pkts.append(pkt)
 294
 295         return pkts
 296
 297     def test_srv6_usid_flex(self):
 298         """ test_srv6_usid_flex """
 299         pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
 300
 301         self.vapi.cli("set sr encaps source addr A1::1")
 302         self.vapi.cli(
 303             "sr localsid prefix 1111:2222:aaaa::/48 behavior un.flex 16")
 304         self.vapi.cli(
 305             "ip route add 1111:2222:bbbb::/48 via {}".format(self.ip6_nhop))
 306
 307         self.logger.info(self.vapi.cli("show sr localsids"))
 308
 309         self.vapi.cli("clear errors")
 310
 311         self.pg0.add_stream(pkts)
 312         self.pg_enable_capture(self.pg_interfaces)
 313         self.pg_start()
 314
 315         self.logger.info(self.vapi.cli("show errors"))
 316         self.logger.info(self.vapi.cli("show int address"))
 317
 318         capture = self.pg1.get_capture(len(pkts))
 319
 320         for pkt in capture:
 321             self.logger.info(pkt.show2(dump=True))
 322             self.assertEqual(
 323                     pkt[IPv6].dst, "1111:2222:bbbb:cccc:dddd:eeee:ffff:0")