6 from scapy.layers.l2 import Ether
7 from scapy.layers.inet import IP, UDP
8 from scapy.packet import Raw
10 from asfframework import VppTestCase, VppTestRunner
13 class TestPcap(VppTestCase):
14 """Pcap Unit Test Cases"""
18 super(TestPcap, cls).setUpClass()
20 cls.create_pg_interfaces(range(1))
21 for i in cls.pg_interfaces:
27 def tearDownClass(cls):
28 for i in cls.pg_interfaces:
31 super(TestPcap, cls).tearDownClass()
34 super(TestPcap, self).setUp()
37 super(TestPcap, self).tearDown()
39 # This is a code coverage test, but it only runs for 0.3 seconds
40 # might as well just run it...
41 def test_pcap_unittest(self):
42 """PCAP Capture Tests"""
45 "set int ip address loop0 11.22.33.1/24",
46 "set int state loop0 up",
48 "set int ip address loop1 11.22.34.1/24",
49 "set int state loop1 up",
50 "set ip neighbor loop1 11.22.34.44 03:00:11:22:34:44",
51 "packet-generator new {\n"
56 " tx-interface loop1\n"
57 " node loop1-output\n"
58 " buffer-flags ip4 offload\n"
59 " buffer-offload-flags offload-ip-cksum offload-udp-cksum\n"
61 " IP4: 1.2.3 -> dead.0000.0001\n"
62 " UDP: 11.22.33.44 -> 11.22.34.44\n"
63 " ttl 2 checksum 13\n"
64 " UDP: 1234 -> 2345\n"
69 "pcap dispatch trace on max 100 buffer-trace pg-input 10",
71 "pcap dispatch trace off",
72 "pcap trace rx tx max 1000 intfc any",
75 "pcap trace rx tx off",
76 "classify filter pcap mask l3 ip4 src match l3 ip4 src 11.22.33.44",
77 "pcap trace rx tx max 1000 intfc any file filt.pcap filter",
78 "show cla t verbose 2",
82 "pcap trace rx tx off",
83 "classify filter pcap del mask l3 ip4 src",
87 r = self.vapi.cli_return_response(cmd)
89 if hasattr(r, "reply"):
90 self.logger.info(cmd + " FAIL reply " + r.reply)
92 self.logger.info(cmd + " FAIL retval " + str(r.retval))
94 self.assertTrue(os.path.exists("/tmp/dispatch.pcap"))
95 self.assertTrue(os.path.exists("/tmp/rxtx.pcap"))
96 self.assertTrue(os.path.exists("/tmp/filt.pcap"))
97 os.remove("/tmp/dispatch.pcap")
98 os.remove("/tmp/rxtx.pcap")
99 os.remove("/tmp/filt.pcap")
101 def test_pcap_trace_api(self):
105 Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
106 / IP(src=self.pg0.local_ip4, dst=self.pg0.remote_ip4, ttl=2)
107 / UDP(sport=1234, dport=2345)
111 self.vapi.pcap_trace_on(
116 filename="trace_any.pcap",
118 self.pg_send(self.pg0, pkt * 10)
119 self.vapi.pcap_trace_off()
122 f"classify filter pcap mask l3 ip4 src match l3 ip4 src {self.pg0.local_ip4}"
124 self.vapi.pcap_trace_on(
130 filename="trace_any_filter.pcap",
132 self.pg_send(self.pg0, pkt * 10)
133 self.vapi.pcap_trace_off()
134 self.vapi.cli("classify filter pcap del mask l3 ip4 src")
137 Ether(src=self.pg0.local_mac, dst=self.pg0.remote_mac)
138 # wrong destination address
139 / IP(src=self.pg0.local_ip4, dst=self.pg0.local_ip4, ttl=2)
140 / UDP(sport=1234, dport=2345)
144 self.vapi.pcap_trace_on(
148 error="{ip4-local}.{spoofed_local_packets}",
149 filename="trace_drop_err.pcap",
151 self.pg_send(self.pg0, pkt * 10)
152 self.vapi.pcap_trace_off()
154 self.assertTrue(os.path.exists("/tmp/trace_any.pcap"))
155 self.assertTrue(os.path.exists("/tmp/trace_any_filter.pcap"))
156 self.assertTrue(os.path.exists("/tmp/trace_drop_err.pcap"))
157 os.remove("/tmp/trace_any.pcap")
158 os.remove("/tmp/trace_any_filter.pcap")
159 os.remove("/tmp/trace_drop_err.pcap")
162 if __name__ == "__main__":
163 unittest.main(testRunner=VppTestRunner)