6 from scapy.layers.l2 import Ether
7 from scapy.layers.inet import IP, UDP
8 from scapy.packet import Raw
10 from framework import VppTestCase
11 from asfframework import VppTestRunner
14 class TestPcap(VppTestCase):
15 """Pcap Unit Test Cases"""
19 super(TestPcap, cls).setUpClass()
21 cls.create_pg_interfaces(range(1))
22 for i in cls.pg_interfaces:
28 def tearDownClass(cls):
29 for i in cls.pg_interfaces:
32 super(TestPcap, cls).tearDownClass()
35 super(TestPcap, self).setUp()
38 super(TestPcap, self).tearDown()
40 # This is a code coverage test, but it only runs for 0.3 seconds
41 # might as well just run it...
42 def test_pcap_unittest(self):
43 """PCAP Capture Tests"""
46 "set int ip address loop0 11.22.33.1/24",
47 "set int state loop0 up",
49 "set int ip address loop1 11.22.34.1/24",
50 "set int state loop1 up",
51 "set ip neighbor loop1 11.22.34.44 03:00:11:22:34:44",
52 "packet-generator new {\n"
57 " tx-interface loop1\n"
58 " node loop1-output\n"
59 " buffer-flags ip4 offload\n"
60 " buffer-offload-flags offload-ip-cksum offload-udp-cksum\n"
62 " IP4: 1.2.3 -> dead.0000.0001\n"
63 " UDP: 11.22.33.44 -> 11.22.34.44\n"
64 " ttl 2 checksum 13\n"
65 " UDP: 1234 -> 2345\n"
70 "pcap dispatch trace on max 100 buffer-trace pg-input 10",
72 "pcap dispatch trace off",
73 "pcap trace rx tx max 1000 intfc any",
76 "pcap trace rx tx off",
77 "classify filter pcap mask l3 ip4 src match l3 ip4 src 11.22.33.44",
78 "pcap trace rx tx max 1000 intfc any file filt.pcap filter",
79 "show cla t verbose 2",
83 "pcap trace rx tx off",
84 "classify filter pcap del mask l3 ip4 src",
88 r = self.vapi.cli_return_response(cmd)
90 if hasattr(r, "reply"):
91 self.logger.info(cmd + " FAIL reply " + r.reply)
93 self.logger.info(cmd + " FAIL retval " + str(r.retval))
95 self.assertTrue(os.path.exists("/tmp/dispatch.pcap"))
96 self.assertTrue(os.path.exists("/tmp/rxtx.pcap"))
97 self.assertTrue(os.path.exists("/tmp/filt.pcap"))
98 os.remove("/tmp/dispatch.pcap")
99 os.remove("/tmp/rxtx.pcap")
100 os.remove("/tmp/filt.pcap")
102 def test_pcap_trace_api(self):
106 Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
107 / IP(src=self.pg0.local_ip4, dst=self.pg0.remote_ip4, ttl=2)
108 / UDP(sport=1234, dport=2345)
112 self.vapi.pcap_trace_on(
117 filename="trace_any.pcap",
119 self.pg_send(self.pg0, pkt * 10)
120 self.vapi.pcap_trace_off()
123 f"classify filter pcap mask l3 ip4 src match l3 ip4 src {self.pg0.local_ip4}"
125 self.vapi.pcap_trace_on(
131 filename="trace_any_filter.pcap",
133 self.pg_send(self.pg0, pkt * 10)
134 self.vapi.pcap_trace_off()
135 self.vapi.cli("classify filter pcap del mask l3 ip4 src")
138 Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
139 # wrong destination address
140 / IP(src=self.pg0.local_ip4, dst=self.pg0.local_ip4, ttl=2)
141 / UDP(sport=1234, dport=2345)
145 self.vapi.pcap_trace_on(
149 error="{ip4-local}.{spoofed_local_packets}",
150 filename="trace_drop_err.pcap",
152 self.pg_send(self.pg0, pkt * 10)
153 self.vapi.pcap_trace_off()
155 self.assertTrue(os.path.exists("/tmp/trace_any.pcap"))
156 self.assertTrue(os.path.exists("/tmp/trace_any_filter.pcap"))
157 self.assertTrue(os.path.exists("/tmp/trace_drop_err.pcap"))
158 os.remove("/tmp/trace_any.pcap")
159 os.remove("/tmp/trace_any_filter.pcap")
160 os.remove("/tmp/trace_drop_err.pcap")
163 if __name__ == "__main__":
164 unittest.main(testRunner=VppTestRunner)