4 from scapy.packet import Packet
5 from scapy.all import ShortField, StrField
6 from scapy.layers.l2 import Dot3, LLC, SNAP
7 from scapy.contrib.cdp import (
15 from framework import VppTestCase
16 from scapy.all import raw
17 from re import compile
18 from time import sleep
25 """ TestCDP is a subclass of VPPTestCase classes.
32 class CustomTLV(Packet):
33 """Custom TLV protocol layer for scapy"""
36 ShortField("type", 0),
37 ShortField("length", 4),
38 StrField("value", ""),
42 class TestCDP(VppTestCase):
45 nen_ptr = compile(r"not enabled")
46 cdp_ptr = compile(r"^([-\.\w]+)\s+([-\.\w]+)\s+([-\.\w]+)\s+([-\.\w]+)$")
47 err_ptr = compile(r"^([\d]+)\s+([-\w]+)\s+([ -\.\w)(]+)$")
51 return platform.node()
55 return platform.release()
59 return self.interface.name
63 return platform.system()
67 super(TestCDP, cls).setUpClass()
69 cls.create_pg_interfaces(range(1))
70 cls.interface = cls.pg_interfaces[0]
72 cls.interface.admin_up()
73 cls.interface.config_ip4()
74 cls.interface.resolve_arp()
77 super(TestCDP, cls).tearDownClass()
81 def tearDownClass(cls):
82 super(TestCDP, cls).tearDownClass()
84 def test_enable_cdp(self):
85 self.logger.info(self.vapi.cdp_enable_disable(enable_disable=1))
86 ret = self.vapi.cli("show cdp")
88 not_enabled = self.nen_ptr.search(ret)
89 self.assertFalse(not_enabled, "CDP isn't enabled")
91 def test_send_cdp_packet(self):
92 self.logger.info(self.vapi.cdp_enable_disable(enable_disable=1))
93 self.send_packet(self.create_packet())
95 neighbors = list(self.show_cdp())
96 self.assertTrue(neighbors, "CDP didn't register neighbor")
98 port, system = neighbors[0]
99 length = min(len(system), len(self.device_id))
101 self.assert_equal(port, self.port_id, "CDP received invalid port id")
103 system[:length], self.device_id[:length], "CDP received invalid device id"
106 def test_cdp_underflow_tlv(self):
107 self.send_bad_packet(3, ".")
109 def test_cdp_overflow_tlv(self):
110 self.send_bad_packet(8, ".")
112 def send_bad_packet(self, l, v):
113 self.logger.info(self.vapi.cdp_enable_disable(enable_disable=1))
114 self.send_packet(self.create_bad_packet(l, v))
116 err = self.statistics.get_err_counter(
117 "/err/cdp-input/cdp packets with bad TLVs"
119 self.assertTrue(err >= 1, "CDP didn't drop bad packet")
121 def send_packet(self, packet):
122 self.logger.debug(ppp("Sending packet:", packet))
123 self.interface.add_stream(packet)
126 def create_base_packet(self):
128 Dot3(src=self.interface.remote_mac, dst="01:00:0c:cc:cc:cc")
129 / LLC(dsap=0xAA, ssap=0xAA, ctrl=0x03)
135 def create_packet(self):
137 self.create_base_packet()
138 / CDPMsgDeviceID(val=self.device_id)
139 / CDPMsgSoftwareVersion(val=self.version)
140 / CDPMsgPortID(iface=self.port_id)
141 / CDPMsgPlatform(val=self.platform)
145 def create_bad_packet(self, tl=4, tv=""):
146 packet = self.create_base_packet() / CustomTLV(type=1, length=tl, value=tv)
149 def process_cli(self, exp, ptr):
150 for line in self.vapi.cli(exp).split("\n")[1:]:
151 m = ptr.match(line.strip())
156 for pack in self.process_cli("show cdp", self.cdp_ptr):
158 port, system, _, _ = pack