test/test_trace_filter.py

   1 #!/usr/bin/env python3
   2
   3 import unittest
   4
   5 from framework import VppTestCase, VppTestRunner
   6 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
   7
   8 from scapy.contrib.geneve import GENEVE
   9 from scapy.packet import Raw
  10 from scapy.layers.l2 import Ether
  11 from scapy.layers.inet import IP, UDP
  12 from scapy.layers.vxlan import VXLAN
  13 from scapy.compat import raw
  14 from scapy.utils import rdpcap
  15
  16
  17 class TestTracefilter(VppTestCase):
  18     """Packet Tracer Filter Test"""
  19
  20     @classmethod
  21     def setUpClass(cls):
  22         super(TestTracefilter, cls).setUpClass()
  23
  24     @classmethod
  25     def tearDownClass(cls):
  26         super(TestTracefilter, cls).tearDownClass()
  27
  28     def setUp(self):
  29         super(TestTracefilter, self).setUp()
  30         self.create_pg_interfaces(range(2))
  31         self.pg0.generate_remote_hosts(11)
  32         for i in self.pg_interfaces:
  33             i.admin_up()
  34             i.config_ip4()
  35             i.resolve_arp()
  36
  37     def tearDown(self):
  38         super(TestTracefilter, self).tearDown()
  39         for i in self.pg_interfaces:
  40             i.unconfig()
  41             i.admin_down()
  42
  43     def cli(self, cmd):
  44         r = self.vapi.cli_return_response(cmd)
  45         if r.retval != 0:
  46             s = (
  47                 "reply '%s'" % r.reply
  48                 if hasattr(r, "reply")
  49                 else "retval '%s'" % r.retval
  50             )
  51             raise RuntimeError("cli command '%s' FAIL with %s" % (cmd, s))
  52         return r
  53
  54     # check number of hits for classifier
  55     def assert_hits(self, n):
  56         r = self.cli("show classify table verbose")
  57         self.assertTrue(r.reply.find("hits %i" % n) != -1)
  58
  59     def add_trace_filter(self, mask, match):
  60         self.cli("classify filter trace mask %s match %s" % (mask, match))
  61         self.cli("clear trace")
  62         self.cli("trace add pg-input 1000 filter")
  63
  64     def del_trace_filters(self):
  65         self.cli("classify filter trace del")
  66         r = self.cli("show classify filter")
  67         s = "packet tracer:                 first table none"
  68         self.assertTrue(r.reply.find(s) != -1)
  69
  70     def del_pcap_filters(self):
  71         self.cli("classify filter pcap del")
  72         r = self.cli("show classify filter")
  73         s = "pcap rx/tx/drop:               first table none"
  74         self.assertTrue(r.reply.find(s) != -1)
  75
  76     def test_basic(self):
  77         """Packet Tracer Filter Test"""
  78         self.add_trace_filter(
  79             "l3 ip4 src", "l3 ip4 src %s" % self.pg0.remote_hosts[5].ip4
  80         )
  81         self.add_trace_filter(
  82             "l3 ip4 proto l4 src_port", "l3 ip4 proto 17 l4 src_port 2345"
  83         )
  84         # the packet we are trying to match
  85         p = list()
  86         for i in range(100):
  87             src = self.pg0.remote_hosts[i % len(self.pg0.remote_hosts)].ip4
  88             p.append(
  89                 (
  90                     Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
  91                     / IP(src=src, dst=self.pg1.remote_ip4)
  92                     / UDP(sport=1234, dport=2345)
  93                     / Raw("\xa5" * 100)
  94                 )
  95             )
  96         for i in range(17):
  97             p.append(
  98                 (
  99                     Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 100                     / IP(src=self.pg0.remote_hosts[0].ip4, dst=self.pg1.remote_ip4)
 101                     / UDP(sport=2345, dport=1234)
 102                     / Raw("\xa5" * 100)
 103                 )
 104             )
 105
 106         self.send_and_expect(self.pg0, p, self.pg1, trace=False)
 107
 108         # Check for 9 and 17 classifier hits, which is the right answer
 109         self.assert_hits(9)
 110         self.assert_hits(17)
 111
 112         self.del_trace_filters()
 113
 114     # install a classify rule, inject traffic and check for hits
 115     def assert_classify(self, mask, match, packets, n=None):
 116         self.add_trace_filter("hex %s" % mask, "hex %s" % match)
 117         self.send_and_expect(self.pg0, packets, self.pg1, trace=False)
 118         self.assert_hits(n if n is not None else len(packets))
 119         self.del_trace_filters()
 120
 121     def test_encap(self):
 122         """Packet Tracer Filter Test with encap"""
 123
 124         # the packet we are trying to match
 125         p = (
 126             Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 127             / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
 128             / UDP()
 129             / VXLAN()
 130             / Ether()
 131             / IP()
 132             / UDP()
 133             / GENEVE(vni=1234)
 134             / Ether()
 135             / IP(src="192.168.4.167")
 136             / UDP()
 137             / Raw("\xa5" * 100)
 138         )
 139
 140         #
 141         # compute filter mask & value
 142         # we compute it by XOR'ing a template packet with a modified packet
 143         # we need to set checksums to 0 to make sure scapy will not recompute
 144         # them
 145         #
 146         tmpl = (
 147             Ether()
 148             / IP(chksum=0)
 149             / UDP(chksum=0)
 150             / VXLAN()
 151             / Ether()
 152             / IP(chksum=0)
 153             / UDP(chksum=0)
 154             / GENEVE(vni=0)
 155             / Ether()
 156             / IP(src="0.0.0.0", chksum=0)
 157         )
 158         ori = raw(tmpl)
 159
 160         # the mask
 161         tmpl[GENEVE].vni = 0xFFFFFF
 162         user = tmpl[GENEVE].payload
 163         user[IP].src = "255.255.255.255"
 164         new = raw(tmpl)
 165         mask = "".join(("{:02x}".format(o ^ n) for o, n in zip(ori, new)))
 166
 167         # this does not match (wrong vni)
 168         tmpl[GENEVE].vni = 1
 169         user = tmpl[GENEVE].payload
 170         user[IP].src = "192.168.4.167"
 171         new = raw(tmpl)
 172         match = "".join(("{:02x}".format(o ^ n) for o, n in zip(ori, new)))
 173         self.assert_classify(mask, match, [p] * 11, 0)
 174
 175         # this must match
 176         tmpl[GENEVE].vni = 1234
 177         new = raw(tmpl)
 178         match = "".join(("{:02x}".format(o ^ n) for o, n in zip(ori, new)))
 179         self.assert_classify(mask, match, [p] * 17)
 180
 181     def test_pcap(self):
 182         """Packet Capture Filter Test"""
 183         self.cli(
 184             "classify filter pcap mask l3 ip4 src match l3 ip4 src %s"
 185             % self.pg0.remote_hosts[5].ip4
 186         )
 187         self.cli(
 188             "classify filter pcap "
 189             "mask l3 ip4 proto l4 src_port "
 190             "match l3 ip4 proto 17 l4 src_port 2345"
 191         )
 192         self.cli(
 193             "pcap trace rx tx max 1000 intfc pg0 "
 194             "file vpp_test_trace_filter_test_pcap.pcap filter"
 195         )
 196         # the packet we are trying to match
 197         p = list()
 198         for i in range(100):
 199             src = self.pg0.remote_hosts[i % len(self.pg0.remote_hosts)].ip4
 200             p.append(
 201                 (
 202                     Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 203                     / IP(src=src, dst=self.pg1.remote_ip4)
 204                     / UDP(sport=1234, dport=2345)
 205                     / Raw("\xa5" * 100)
 206                 )
 207             )
 208         for i in range(17):
 209             p.append(
 210                 (
 211                     Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 212                     / IP(src=self.pg0.remote_hosts[0].ip4, dst=self.pg1.remote_ip4)
 213                     / UDP(sport=2345, dport=1234)
 214                     / Raw("\xa5" * 100)
 215                 )
 216             )
 217
 218         self.send_and_expect(self.pg0, p, self.pg1, trace=False)
 219
 220         # Check for 9 and 17 classifier hits, which is the right answer
 221         self.assert_hits(9)
 222         self.assert_hits(17)
 223
 224         self.cli("pcap trace rx tx off")
 225         self.del_pcap_filters()
 226
 227         # check captured pcap
 228         pcap = rdpcap("/tmp/vpp_test_trace_filter_test_pcap.pcap")
 229         self.assertEqual(len(pcap), 9 + 17)
 230         p_ = str(p[5])
 231         for i in range(9):
 232             self.assertEqual(str(pcap[i]), p_)
 233         p_ = str(p[100])
 234         for i in range(9, 9 + 17):
 235             self.assertEqual(str(pcap[i]), p_)
 236
 237     def test_pcap_drop(self):
 238         """Drop Packet Capture Filter Test"""
 239         self.cli(
 240             "pcap trace drop max 1000 "
 241             "error {ip4-udp-lookup}.{no_listener} "
 242             "file vpp_test_trace_filter_test_pcap_drop.pcap"
 243         )
 244         # the packet we are trying to match
 245         p = list()
 246         for i in range(17):
 247             # this packet should be forwarded
 248             p.append(
 249                 (
 250                     Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 251                     / IP(src=self.pg0.remote_hosts[0].ip4, dst=self.pg1.remote_ip4)
 252                     / UDP(sport=2345, dport=1234)
 253                     / Raw("\xa5" * 100)
 254                 )
 255             )
 256             # this packet should be captured (no listener)
 257             p.append(
 258                 (
 259                     Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 260                     / IP(src=self.pg0.remote_hosts[0].ip4, dst=self.pg0.local_ip4)
 261                     / UDP(sport=2345, dport=1234)
 262                     / Raw("\xa5" * 100)
 263                 )
 264             )
 265         # this packet will be blackholed but not captured
 266         p.append(
 267             (
 268                 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 269                 / IP(src=self.pg0.remote_hosts[0].ip4, dst="0.0.0.0")
 270                 / UDP(sport=2345, dport=1234)
 271                 / Raw("\xa5" * 100)
 272             )
 273         )
 274
 275         self.send_and_expect(self.pg0, p, self.pg1, n_rx=17, trace=False)
 276
 277         self.cli("pcap trace drop off")
 278
 279         # check captured pcap
 280         pcap = rdpcap("/tmp/vpp_test_trace_filter_test_pcap_drop.pcap")
 281         self.assertEqual(len(pcap), 17)
 282
 283
 284 if __name__ == "__main__":
 285     unittest.main(testRunner=VppTestRunner)