test/test_srv6_un.py

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