5 from framework import VppTestCase, VppTestRunner
6 from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
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
17 class TestTracefilter(VppTestCase):
18 """Packet Tracer Filter Test"""
22 super(TestTracefilter, cls).setUpClass()
25 def tearDownClass(cls):
26 super(TestTracefilter, cls).tearDownClass()
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:
38 super(TestTracefilter, self).tearDown()
39 for i in self.pg_interfaces:
44 r = self.vapi.cli_return_response(cmd)
47 "reply '%s'" % r.reply
48 if hasattr(r, "reply")
49 else "retval '%s'" % r.retval
51 raise RuntimeError("cli command '%s' FAIL with %s" % (cmd, s))
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)
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")
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)
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)
77 """Packet Tracer Filter Test"""
78 self.add_trace_filter(
79 "l3 ip4 src", "l3 ip4 src %s" % self.pg0.remote_hosts[5].ip4
81 self.add_trace_filter(
82 "l3 ip4 proto l4 src_port", "l3 ip4 proto 17 l4 src_port 2345"
84 # the packet we are trying to match
87 src = self.pg0.remote_hosts[i % len(self.pg0.remote_hosts)].ip4
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)
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)
106 self.send_and_expect(self.pg0, p, self.pg1, trace=False)
108 # Check for 9 and 17 classifier hits, which is the right answer
112 self.del_trace_filters()
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()
121 def test_encap(self):
122 """Packet Tracer Filter Test with encap"""
124 # the packet we are trying to match
126 Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
127 / IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
135 / IP(src="192.168.4.167")
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
156 / IP(src="0.0.0.0", chksum=0)
161 tmpl[GENEVE].vni = 0xFFFFFF
162 user = tmpl[GENEVE].payload
163 user[IP].src = "255.255.255.255"
165 mask = "".join(("{:02x}".format(o ^ n) for o, n in zip(ori, new)))
167 # this does not match (wrong vni)
169 user = tmpl[GENEVE].payload
170 user[IP].src = "192.168.4.167"
172 match = "".join(("{:02x}".format(o ^ n) for o, n in zip(ori, new)))
173 self.assert_classify(mask, match, [p] * 11, 0)
176 tmpl[GENEVE].vni = 1234
178 match = "".join(("{:02x}".format(o ^ n) for o, n in zip(ori, new)))
179 self.assert_classify(mask, match, [p] * 17)
182 """Packet Capture Filter Test"""
184 "classify filter pcap mask l3 ip4 src match l3 ip4 src %s"
185 % self.pg0.remote_hosts[5].ip4
188 "classify filter pcap "
189 "mask l3 ip4 proto l4 src_port "
190 "match l3 ip4 proto 17 l4 src_port 2345"
193 "pcap trace rx tx max 1000 intfc pg0 "
194 "file vpp_test_trace_filter_test_pcap.pcap filter"
196 # the packet we are trying to match
199 src = self.pg0.remote_hosts[i % len(self.pg0.remote_hosts)].ip4
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)
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)
218 self.send_and_expect(self.pg0, p, self.pg1, trace=False)
220 # Check for 9 and 17 classifier hits, which is the right answer
224 self.cli("pcap trace rx tx off")
225 self.del_pcap_filters()
227 # check captured pcap
228 pcap = rdpcap("/tmp/vpp_test_trace_filter_test_pcap.pcap")
229 self.assertEqual(len(pcap), 9 + 17)
232 self.assertEqual(str(pcap[i]), p_)
234 for i in range(9, 9 + 17):
235 self.assertEqual(str(pcap[i]), p_)
237 def test_pcap_drop(self):
238 """Drop Packet Capture Filter Test"""
240 "pcap trace drop max 1000 "
241 "error {ip4-udp-lookup}.{no_listener} "
242 "file vpp_test_trace_filter_test_pcap_drop.pcap"
244 # the packet we are trying to match
247 # this packet should be forwarded
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)
256 # this packet should be captured (no listener)
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)
265 # this packet will be blackholed but not captured
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)
275 self.send_and_expect(self.pg0, p, self.pg1, n_rx=17, trace=False)
277 self.cli("pcap trace drop off")
279 # check captured pcap
280 pcap = rdpcap("/tmp/vpp_test_trace_filter_test_pcap_drop.pcap")
281 self.assertEqual(len(pcap), 17)
284 if __name__ == "__main__":
285 unittest.main(testRunner=VppTestRunner)