5 from scapy.layers.l2 import Ether
6 from scapy.packet import Raw
7 from scapy.layers.inet import IP, UDP
8 from scapy.layers.inet6 import IPv6
9 from scapy.contrib.mpls import MPLS
12 class BridgeDomain(metaclass=abc.ABCMeta):
13 """Bridge domain abstraction"""
16 def frame_request(self):
17 """Ethernet frame modeling a generic request"""
19 Ether(src="00:00:00:00:00:01", dst="00:00:00:00:00:02")
20 / IP(src="1.2.3.4", dst="4.3.2.1")
21 / UDP(sport=10000, dport=20000)
26 def frame_reply(self):
27 """Ethernet frame modeling a generic reply"""
29 Ether(src="00:00:00:00:00:02", dst="00:00:00:00:00:01")
30 / IP(src="4.3.2.1", dst="1.2.3.4")
31 / UDP(sport=20000, dport=10000)
36 def ip_range(self, start, end):
37 """range of remote ip's"""
41 def encap_mcast(self, pkt, src_ip, src_mac, vni):
42 """Encapsulate mcast packet"""
46 def encapsulate(self, pkt, vni):
47 """Encapsulate packet"""
51 def decapsulate(self, pkt):
52 """Decapsulate packet"""
56 def check_encapsulation(self, pkt, vni, local_only=False):
57 """Verify the encapsulation"""
60 def assert_eq_pkts(self, pkt1, pkt2):
61 """Verify the Ether, IP, UDP, payload are equal in both
64 self.assertEqual(pkt1[Ether].src, pkt2[Ether].src)
65 self.assertEqual(pkt1[Ether].dst, pkt2[Ether].dst)
66 if MPLS in pkt1 or MPLS in pkt2:
67 self.assertEqual(pkt1[MPLS].label, pkt2[MPLS].label)
68 self.assertEqual(pkt1[MPLS].cos, pkt2[MPLS].cos)
69 self.assertEqual(pkt1[MPLS].ttl, pkt2[MPLS].ttl)
70 if IP in pkt1 or IP in pkt2:
71 self.assertEqual(pkt1[IP].src, pkt2[IP].src)
72 self.assertEqual(pkt1[IP].dst, pkt2[IP].dst)
73 elif IPv6 in pkt1 or IPv6 in pkt2:
74 self.assertEqual(pkt1[IPv6].src, pkt2[IPv6].src)
75 self.assertEqual(pkt1[IPv6].dst, pkt2[IPv6].dst)
76 self.assertEqual(pkt1[UDP].sport, pkt2[UDP].sport)
77 self.assertEqual(pkt1[UDP].dport, pkt2[UDP].dport)
78 self.assertEqual(pkt1[Raw], pkt2[Raw])
82 Send encapsulated frames from pg0
83 Verify receipt of decapsulated frames on pg1
86 encapsulated_pkt = self.encapsulate(self.frame_request, self.single_tunnel_vni)
94 self.pg1.enable_capture()
98 # Pick first received frame and check if it's the non-encapsulated
100 out = self.pg1.get_capture(1)
102 self.assert_eq_pkts(pkt, self.frame_request)
104 def test_encap(self):
105 """Encapsulation test
107 Verify receipt of encapsulated frames on pg0
109 self.pg1.add_stream([self.frame_reply])
111 self.pg0.enable_capture()
115 # Pick first received frame and check if it's correctly encapsulated.
116 out = self.pg0.get_capture(1)
118 self.check_encapsulation(pkt, self.single_tunnel_vni)
120 payload = self.decapsulate(pkt)
121 self.assert_eq_pkts(payload, self.frame_reply)
123 def test_ucast_flood(self):
124 """Unicast flood test
126 Verify receipt of encapsulated frames on pg0
128 self.pg3.add_stream([self.frame_reply])
130 self.pg0.enable_capture()
134 # Get packet from each tunnel and assert it's correctly encapsulated.
135 out = self.pg0.get_capture(self.n_ucast_tunnels)
137 self.check_encapsulation(pkt, self.ucast_flood_bd, True)
138 payload = self.decapsulate(pkt)
139 self.assert_eq_pkts(payload, self.frame_reply)
141 def test_mcast_flood(self):
142 """Multicast flood test
144 Verify receipt of encapsulated frames on pg0
146 self.pg2.add_stream([self.frame_reply])
148 self.pg0.enable_capture()
152 # Pick first received frame and check if it's correctly encapsulated.
153 out = self.pg0.get_capture(1)
155 self.check_encapsulation(
156 pkt, self.mcast_flood_bd, local_only=False, mcast_pkt=True
159 payload = self.decapsulate(pkt)
160 self.assert_eq_pkts(payload, self.frame_reply)
162 def test_mcast_rcv(self):
163 """Multicast receive test
164 Send 20 encapsulated frames from pg0 only 10 match unicast tunnels
165 Verify receipt of 10 decap frames on pg2
167 mac = self.pg0.remote_mac
171 self.encap_mcast(self.frame_request, ip, mac, self.mcast_flood_bd)
172 for ip in self.ip_range(ip_range_start, ip_range_end)
174 self.pg0.add_stream(mcast_stream)
175 self.pg2.enable_capture()
177 out = self.pg2.get_capture(10)
179 self.assert_eq_pkts(pkt, self.frame_request)