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 *
10 class TestSRv6EndMGTP4E(VppTestCase):
11 """ SRv6 End.M.GTP4.E (SRv6 -> GTP-U) """
15 super(TestSRv6EndMGTP4E, cls).setUpClass()
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]
21 cls.pg_if_i.config_ip6()
22 cls.pg_if_o.config_ip4()
24 cls.ip4_dst = cls.pg_if_o.remote_ip4
25 # cls.ip4_src = cls.pg_if_o.local_ip4
26 cls.ip4_src = "192.168.192.10"
28 for pg_if in cls.pg_interfaces:
33 super(TestSRv6EndMGTP4E, cls).tearDownClass()
36 def create_packets(self, inner):
38 ip4_dst = IPv4Address(str(self.ip4_dst))
39 # 32bit prefix + 32bit IPv4 DA + 8bit + 32bit TEID + 24bit
40 dst = b'\xaa' * 4 + ip4_dst.packed + \
41 b'\x11' + b'\xbb' * 4 + b'\x11' * 3
42 ip6_dst = IPv6Address(dst)
44 ip4_src = IPv4Address(str(self.ip4_src))
45 # 64bit prefix + 32bit IPv4 SA + 16 bit port + 16bit
46 src = b'\xcc' * 8 + ip4_src.packed + \
47 b'\xdd' * 2 + b'\x11' * 2
48 ip6_src = IPv6Address(src)
50 self.logger.info("ip4 dst: {}".format(ip4_dst))
51 self.logger.info("ip4 src: {}".format(ip4_src))
52 self.logger.info("ip6 dst (remote srgw): {}".format(ip6_dst))
53 self.logger.info("ip6 src (local srgw): {}".format(ip6_src))
58 IPv6(dst=str(ip6_dst), src=str(ip6_src)) /
59 IPv6ExtHdrSegmentRouting() /
61 UDP(sport=1000, dport=23))
62 self.logger.info(pkt.show2(dump=True))
67 def test_srv6_mobile(self):
68 """ test_srv6_mobile """
69 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
72 "sr localsid address {} behavior end.m.gtp4.e v4src_position 64"
73 .format(pkts[0]['IPv6'].dst))
74 self.logger.info(self.vapi.cli("show sr localsids"))
76 self.vapi.cli("clear errors")
78 self.pg0.add_stream(pkts)
79 self.pg_enable_capture(self.pg_interfaces)
82 self.logger.info(self.vapi.cli("show errors"))
83 self.logger.info(self.vapi.cli("show int address"))
85 capture = self.pg1.get_capture(len(pkts))
88 self.logger.info(pkt.show2(dump=True))
89 self.assertEqual(pkt[IP].dst, self.ip4_dst)
90 self.assertEqual(pkt[IP].src, self.ip4_src)
91 self.assertEqual(pkt[GTP_U_Header].teid, 0xbbbbbbbb)
94 class TestSRv6TMGTP4D(VppTestCase):
95 """ SRv6 T.M.GTP4.D (GTP-U -> SRv6) """
99 super(TestSRv6TMGTP4D, cls).setUpClass()
101 cls.create_pg_interfaces(range(2))
102 cls.pg_if_i = cls.pg_interfaces[0]
103 cls.pg_if_o = cls.pg_interfaces[1]
105 cls.pg_if_i.config_ip4()
106 cls.pg_if_i.config_ip6()
107 cls.pg_if_o.config_ip4()
108 cls.pg_if_o.config_ip6()
110 cls.ip4_dst = "1.1.1.1"
111 cls.ip4_src = "2.2.2.2"
113 cls.ip6_dst = cls.pg_if_o.remote_ip6
115 for pg_if in cls.pg_interfaces:
118 pg_if.resolve_ndp(timeout=5)
121 super(TestSRv6TMGTP4D, cls).tearDownClass()
124 def create_packets(self, inner):
126 ip4_dst = IPv4Address(str(self.ip4_dst))
128 ip4_src = IPv4Address(str(self.ip4_src))
130 self.logger.info("ip4 dst: {}".format(ip4_dst))
131 self.logger.info("ip4 src: {}".format(ip4_src))
136 IP(dst=str(ip4_dst), src=str(ip4_src)) /
137 UDP(sport=2152, dport=2152) /
138 GTP_U_Header(gtp_type="g_pdu", teid=200) /
140 UDP(sport=1000, dport=23))
141 self.logger.info(pkt.show2(dump=True))
146 def test_srv6_mobile(self):
147 """ test_srv6_mobile """
148 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
150 self.vapi.cli("set sr encaps source addr A1::1")
151 self.vapi.cli("sr policy add bsid D4:: next D2:: next D3::")
153 "sr policy add bsid D5:: behavior t.m.gtp4.d"
154 "D4::/32 v6src_prefix C1::/64 nhtype ipv6")
155 self.vapi.cli("sr steer l3 {}/32 via bsid D5::".format(self.ip4_dst))
156 self.vapi.cli("ip route add D2::/32 via {}".format(self.ip6_dst))
158 self.logger.info(self.vapi.cli("show sr steer"))
159 self.logger.info(self.vapi.cli("show sr policies"))
161 self.vapi.cli("clear errors")
163 self.pg0.add_stream(pkts)
164 self.pg_enable_capture(self.pg_interfaces)
167 self.logger.info(self.vapi.cli("show errors"))
168 self.logger.info(self.vapi.cli("show int address"))
170 capture = self.pg1.get_capture(len(pkts))
173 self.logger.info(pkt.show2(dump=True))
174 self.logger.info("GTP4.D Address={}".format(
175 str(pkt[IPv6ExtHdrSegmentRouting].addresses[0])))
177 str(pkt[IPv6ExtHdrSegmentRouting].addresses[0]),
178 "d4:0:101:101::c800:0")
181 class TestSRv6EndMGTP6E(VppTestCase):
182 """ SRv6 End.M.GTP6.E """
186 super(TestSRv6EndMGTP6E, cls).setUpClass()
188 cls.create_pg_interfaces(range(2))
189 cls.pg_if_i = cls.pg_interfaces[0]
190 cls.pg_if_o = cls.pg_interfaces[1]
192 cls.pg_if_i.config_ip6()
193 cls.pg_if_o.config_ip6()
195 cls.ip6_nhop = cls.pg_if_o.remote_ip6
197 for pg_if in cls.pg_interfaces:
199 pg_if.resolve_ndp(timeout=5)
202 super(TestSRv6EndMGTP6E, cls).tearDownClass()
205 def create_packets(self, inner):
206 # 64bit prefix + 8bit QFI + 32bit TEID + 24bit
207 dst = b'\xaa' * 8 + b'\x00' + \
208 b'\xbb' * 4 + b'\x00' * 3
209 ip6_dst = IPv6Address(dst)
211 self.ip6_dst = ip6_dst
213 src = b'\xcc' * 8 + \
214 b'\xdd' * 4 + b'\x11' * 4
215 ip6_src = IPv6Address(src)
217 self.ip6_src = ip6_src
222 IPv6(dst=str(ip6_dst),
224 IPv6ExtHdrSegmentRouting(segleft=1,
227 addresses=["a1::1"]) /
228 IPv6(dst=d, src=s) / UDP(sport=1000, dport=23))
229 self.logger.info(pkt.show2(dump=True))
234 def test_srv6_mobile(self):
235 """ test_srv6_mobile """
236 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
239 "sr localsid prefix {}/64 behavior end.m.gtp6.e"
240 .format(pkts[0]['IPv6'].dst))
242 "ip route add a1::/64 via {}".format(self.ip6_nhop))
243 self.logger.info(self.vapi.cli("show sr localsids"))
245 self.vapi.cli("clear errors")
247 self.pg0.add_stream(pkts)
248 self.pg_enable_capture(self.pg_interfaces)
251 self.logger.info(self.vapi.cli("show errors"))
252 self.logger.info(self.vapi.cli("show int address"))
254 capture = self.pg1.get_capture(len(pkts))
257 self.logger.info(pkt.show2(dump=True))
258 self.assertEqual(pkt[IPv6].dst, "a1::1")
259 self.assertEqual(pkt[IPv6].src, str(self.ip6_src))
260 self.assertEqual(pkt[GTP_U_Header].teid, 0xbbbbbbbb)
263 class TestSRv6EndMGTP6D(VppTestCase):
264 """ SRv6 End.M.GTP6.D """
268 super(TestSRv6EndMGTP6D, cls).setUpClass()
270 cls.create_pg_interfaces(range(2))
271 cls.pg_if_i = cls.pg_interfaces[0]
272 cls.pg_if_o = cls.pg_interfaces[1]
274 cls.pg_if_i.config_ip6()
275 cls.pg_if_o.config_ip6()
277 cls.ip6_nhop = cls.pg_if_o.remote_ip6
279 cls.ip6_dst = "2001::1"
280 cls.ip6_src = "2002::1"
282 for pg_if in cls.pg_interfaces:
284 pg_if.resolve_ndp(timeout=5)
287 super(TestSRv6EndMGTP6D, cls).tearDownClass()
290 def create_packets(self, inner):
292 ip6_dst = IPv6Address(str(self.ip6_dst))
294 ip6_src = IPv6Address(str(self.ip6_src))
296 self.logger.info("ip6 dst: {}".format(ip6_dst))
297 self.logger.info("ip6 src: {}".format(ip6_src))
302 IPv6(dst=str(ip6_dst), src=str(ip6_src)) /
303 UDP(sport=2152, dport=2152) /
304 GTP_U_Header(gtp_type="g_pdu", teid=200) /
306 UDP(sport=1000, dport=23))
307 self.logger.info(pkt.show2(dump=True))
312 def test_srv6_mobile(self):
313 """ test_srv6_mobile """
314 pkts = self.create_packets([("A::1", "B::1"), ("C::1", "D::1")])
316 self.vapi.cli("set sr encaps source addr A1::1")
317 self.vapi.cli("sr policy add bsid D4:: next D2:: next D3::")
319 "sr localsid prefix 2001::/64 behavior end.m.gtp6.d D4::/64")
320 self.vapi.cli("ip route add D2::/64 via {}".format(self.ip6_nhop))
322 self.logger.info(self.vapi.cli("show sr policies"))
324 self.vapi.cli("clear errors")
326 self.pg0.add_stream(pkts)
327 self.pg_enable_capture(self.pg_interfaces)
330 self.logger.info(self.vapi.cli("show errors"))
331 self.logger.info(self.vapi.cli("show int address"))
333 capture = self.pg1.get_capture(len(pkts))
336 self.logger.info(pkt.show2(dump=True))
337 self.logger.info("GTP6.D Address={}".format(
338 str(pkt[IPv6ExtHdrSegmentRouting].addresses[0])))
340 str(pkt[IPv6ExtHdrSegmentRouting].addresses[0]), "d4::c800:0")