test/template_bd.py

   1 #!/usr/bin/env python3
   2
   3 import abc
   4
   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
  10
  11
  12 class BridgeDomain(metaclass=abc.ABCMeta):
  13     """Bridge domain abstraction"""
  14
  15     @property
  16     def frame_request(self):
  17         """Ethernet frame modeling a generic request"""
  18         return (
  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)
  22             / Raw("\xa5" * 100)
  23         )
  24
  25     @property
  26     def frame_reply(self):
  27         """Ethernet frame modeling a generic reply"""
  28         return (
  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)
  32             / Raw("\xa5" * 100)
  33         )
  34
  35     @abc.abstractmethod
  36     def ip_range(self, start, end):
  37         """range of remote ip's"""
  38         pass
  39
  40     @abc.abstractmethod
  41     def encap_mcast(self, pkt, src_ip, src_mac, vni):
  42         """Encapsulate mcast packet"""
  43         pass
  44
  45     @abc.abstractmethod
  46     def encapsulate(self, pkt, vni):
  47         """Encapsulate packet"""
  48         pass
  49
  50     @abc.abstractmethod
  51     def decapsulate(self, pkt):
  52         """Decapsulate packet"""
  53         pass
  54
  55     @abc.abstractmethod
  56     def check_encapsulation(self, pkt, vni, local_only=False):
  57         """Verify the encapsulation"""
  58         pass
  59
  60     def assert_eq_pkts(self, pkt1, pkt2):
  61         """Verify the Ether, IP, UDP, payload are equal in both
  62         packets
  63         """
  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])
  79
  80     def test_decap(self):
  81         """Decapsulation test
  82         Send encapsulated frames from pg0
  83         Verify receipt of decapsulated frames on pg1
  84         """
  85
  86         encapsulated_pkt = self.encapsulate(self.frame_request, self.single_tunnel_vni)
  87
  88         self.pg0.add_stream(
  89             [
  90                 encapsulated_pkt,
  91             ]
  92         )
  93
  94         self.pg1.enable_capture()
  95
  96         self.pg_start()
  97
  98         # Pick first received frame and check if it's the non-encapsulated
  99         # frame
 100         out = self.pg1.get_capture(1)
 101         pkt = out[0]
 102         self.assert_eq_pkts(pkt, self.frame_request)
 103
 104     def test_encap(self):
 105         """Encapsulation test
 106         Send frames from pg1
 107         Verify receipt of encapsulated frames on pg0
 108         """
 109         self.pg1.add_stream([self.frame_reply])
 110
 111         self.pg0.enable_capture()
 112
 113         self.pg_start()
 114
 115         # Pick first received frame and check if it's correctly encapsulated.
 116         out = self.pg0.get_capture(1)
 117         pkt = out[0]
 118         self.check_encapsulation(pkt, self.single_tunnel_vni)
 119
 120         payload = self.decapsulate(pkt)
 121         self.assert_eq_pkts(payload, self.frame_reply)
 122
 123     def test_ucast_flood(self):
 124         """Unicast flood test
 125         Send frames from pg3
 126         Verify receipt of encapsulated frames on pg0
 127         """
 128         self.pg3.add_stream([self.frame_reply])
 129
 130         self.pg0.enable_capture()
 131
 132         self.pg_start()
 133
 134         # Get packet from each tunnel and assert it's correctly encapsulated.
 135         out = self.pg0.get_capture(self.n_ucast_tunnels)
 136         for pkt in out:
 137             self.check_encapsulation(pkt, self.ucast_flood_bd, True)
 138             payload = self.decapsulate(pkt)
 139             self.assert_eq_pkts(payload, self.frame_reply)
 140
 141     def test_mcast_flood(self):
 142         """Multicast flood test
 143         Send frames from pg2
 144         Verify receipt of encapsulated frames on pg0
 145         """
 146         self.pg2.add_stream([self.frame_reply])
 147
 148         self.pg0.enable_capture()
 149
 150         self.pg_start()
 151
 152         # Pick first received frame and check if it's correctly encapsulated.
 153         out = self.pg0.get_capture(1)
 154         pkt = out[0]
 155         self.check_encapsulation(
 156             pkt, self.mcast_flood_bd, local_only=False, mcast_pkt=True
 157         )
 158
 159         payload = self.decapsulate(pkt)
 160         self.assert_eq_pkts(payload, self.frame_reply)
 161
 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
 166         """
 167         mac = self.pg0.remote_mac
 168         ip_range_start = 10
 169         ip_range_end = 30
 170         mcast_stream = [
 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)
 173         ]
 174         self.pg0.add_stream(mcast_stream)
 175         self.pg2.enable_capture()
 176         self.pg_start()
 177         out = self.pg2.get_capture(10)
 178         for pkt in out:
 179             self.assert_eq_pkts(pkt, self.frame_request)