test/test_srv6_un.py

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