test/test_pcap.py

   1 #!/usr/bin/env python3
   2
   3 import os
   4 import unittest
   5
   6 from scapy.layers.l2 import Ether
   7 from scapy.layers.inet import IP, UDP
   8 from scapy.packet import Raw
   9
  10 from framework import VppTestCase
  11 from asfframework import VppTestRunner
  12
  13
  14 class TestPcap(VppTestCase):
  15     """Pcap Unit Test Cases"""
  16
  17     @classmethod
  18     def setUpClass(cls):
  19         super(TestPcap, cls).setUpClass()
  20
  21         cls.create_pg_interfaces(range(1))
  22         for i in cls.pg_interfaces:
  23             i.admin_up()
  24             i.config_ip4()
  25             i.resolve_arp()
  26
  27     @classmethod
  28     def tearDownClass(cls):
  29         for i in cls.pg_interfaces:
  30             i.admin_down()
  31
  32         super(TestPcap, cls).tearDownClass()
  33
  34     def setUp(self):
  35         super(TestPcap, self).setUp()
  36
  37     def tearDown(self):
  38         super(TestPcap, self).tearDown()
  39
  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"""
  44         cmds = [
  45             "loop create",
  46             "set int ip address loop0 11.22.33.1/24",
  47             "set int state loop0 up",
  48             "loop create",
  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"
  53             "  name s0\n"
  54             "  limit 10\n"
  55             "  size 128-128\n"
  56             "  interface loop0\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"
  61             "  data {\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"
  66             "      checksum 11\n"
  67             "    incrementing 114\n"
  68             "  }\n"
  69             "}",
  70             "pcap dispatch trace on max 100 buffer-trace pg-input 10",
  71             "pa en",
  72             "pcap dispatch trace off",
  73             "pcap trace rx tx max 1000 intfc any",
  74             "pa en",
  75             "pcap trace status",
  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",
  80             "show cla t verbose",
  81             "show cla t",
  82             "pa en",
  83             "pcap trace rx tx off",
  84             "classify filter pcap del mask l3 ip4 src",
  85         ]
  86
  87         for cmd in cmds:
  88             r = self.vapi.cli_return_response(cmd)
  89             if r.retval != 0:
  90                 if hasattr(r, "reply"):
  91                     self.logger.info(cmd + " FAIL reply " + r.reply)
  92                 else:
  93                     self.logger.info(cmd + " FAIL retval " + str(r.retval))
  94
  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")
 101
 102     def test_pcap_trace_api(self):
 103         """PCAP API Tests"""
 104
 105         pkt = (
 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)
 109             / Raw(b"\xa5" * 128)
 110         )
 111
 112         self.vapi.pcap_trace_on(
 113             capture_rx=True,
 114             capture_tx=True,
 115             max_packets=1000,
 116             sw_if_index=0,
 117             filename="trace_any.pcap",
 118         )
 119         self.pg_send(self.pg0, pkt * 10)
 120         self.vapi.pcap_trace_off()
 121
 122         self.vapi.cli(
 123             f"classify filter pcap mask l3 ip4 src match l3 ip4 src {self.pg0.local_ip4}"
 124         )
 125         self.vapi.pcap_trace_on(
 126             capture_rx=True,
 127             capture_tx=True,
 128             filter=True,
 129             max_packets=1000,
 130             sw_if_index=0,
 131             filename="trace_any_filter.pcap",
 132         )
 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")
 136
 137         pkt = (
 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)
 142             / Raw(b"\xa5" * 128)
 143         )
 144
 145         self.vapi.pcap_trace_on(
 146             capture_drop=True,
 147             max_packets=1000,
 148             sw_if_index=0,
 149             error="{ip4-local}.{spoofed_local_packets}",
 150             filename="trace_drop_err.pcap",
 151         )
 152         self.pg_send(self.pg0, pkt * 10)
 153         self.vapi.pcap_trace_off()
 154
 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")
 161
 162
 163 if __name__ == "__main__":
 164     unittest.main(testRunner=VppTestRunner)