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