from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, TCP, UDP
from scapy.layers.inet6 import IPv6
+from scapy.contrib.lacp import SlowProtocol, LACP
from config import config
-from framework import tag_fixme_vpp_workers
+from framework import tag_fixme_vpp_workers, tag_fixme_ubuntu2204, tag_fixme_debian11
+from framework import is_distro_ubuntu2204, is_distro_debian11
from framework import VppTestCase, VppTestRunner
from framework import tag_run_solo
from vpp_object import VppObject
from vpp_papi import VppEnum
+TMPL_COMMON_FIELD_COUNT = 6
+TMPL_L2_FIELD_COUNT = 3
+TMPL_L3_FIELD_COUNT = 4
+TMPL_L4_FIELD_COUNT = 3
+
+IPFIX_TCP_FLAGS_ID = 6
+IPFIX_SRC_TRANS_PORT_ID = 7
+IPFIX_DST_TRANS_PORT_ID = 11
+
+TCP_F_FIN = 0x01
+TCP_F_SYN = 0x02
+TCP_F_RST = 0x04
+TCP_F_PSH = 0x08
+TCP_F_ACK = 0x10
+TCP_F_URG = 0x20
+TCP_F_ECE = 0x40
+TCP_F_CWR = 0x80
+
+
class VppCFLOW(VppObject):
"""CFLOW object for IPFIX exporter and Flowprobe feature"""
def query_vpp_config(self):
return self._configured
- def verify_templates(self, decoder=None, timeout=1, count=3):
+ def verify_templates(self, decoder=None, timeout=1, count=3, field_count_in=None):
templates = []
self._test.assertIn(count, (1, 2, 3))
for _ in range(count):
p = self._test.wait_for_cflow_packet(self._test.collector, 2, timeout)
self._test.assertTrue(p.haslayer(IPFIX))
- if decoder is not None and p.haslayer(Template):
+ self._test.assertTrue(p.haslayer(Template))
+ if decoder is not None:
templates.append(p[Template].templateID)
decoder.add_template(p.getlayer(Template))
+ if field_count_in is not None:
+ self._test.assertIn(p[Template].fieldCount, field_count_in)
return templates
variables and configure VPP.
"""
super(MethodHolder, cls).setUpClass()
+ if (is_distro_ubuntu2204 == True or is_distro_debian11 == True) and not hasattr(
+ cls, "vpp"
+ ):
+ return
try:
# Create pg interfaces
cls.create_pg_interfaces(range(9))
# Create BD with MAC learning and unknown unicast flooding disabled
# and put interfaces to this BD
- cls.vapi.bridge_domain_add_del(bd_id=1, uu_flood=1, learn=1)
+ cls.vapi.bridge_domain_add_del_v2(
+ bd_id=1, uu_flood=1, learn=1, flood=1, forward=1, is_add=1
+ )
cls.vapi.sw_interface_set_l2_bridge(
rx_sw_if_index=cls.pg1._sw_if_index, bd_id=1
)
return dst_if.get_capture(len(self.pkts))
def verify_cflow_data_detail(
- self, decoder, capture, cflow, data_set={1: "octets", 2: "packets"}, ip_ver="v4"
+ self,
+ decoder,
+ capture,
+ cflow,
+ data_set={1: "octets", 2: "packets"},
+ ip_ver="v4",
+ field_count=None,
):
if self.debug_print:
print(capture[0].show())
if self.debug_print:
print(data)
if ip_ver == "v4":
- ip_layer = capture[0][IP]
+ ip_layer = capture[0][IP] if capture[0].haslayer(IP) else None
else:
- ip_layer = capture[0][IPv6]
+ ip_layer = capture[0][IPv6] if capture[0].haslayer(IPv6) else None
if data_set is not None:
for record in data:
# skip flow if ingress/egress interface is 0
self.assertEqual(
int(binascii.hexlify(record[field]), 16), value
)
+ if field_count is not None:
+ for record in data:
+ self.assertEqual(len(record), field_count)
def verify_cflow_data_notimer(self, decoder, capture, cflows):
idx = 0
@tag_run_solo
@tag_fixme_vpp_workers
+@tag_fixme_ubuntu2204
+@tag_fixme_debian11
class Flowprobe(MethodHolder):
"""Template verification, timer tests"""
ipfix.remove_vpp_config()
self.logger.info("FFP_TEST_FINISH_0000")
+ def test_flow_entry_reuse(self):
+ """Verify flow entry reuse doesn't accumulate meta info"""
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pkts = []
+
+ # enable ip4 datapath for an interface
+ # set active and passive timers
+ ipfix = VppCFLOW(
+ test=self,
+ active=2,
+ passive=3,
+ intf="pg3",
+ layer="l3 l4",
+ datapath="ip4",
+ direction="rx",
+ mtu=100,
+ )
+ ipfix.add_vpp_config()
+
+ # template packet should arrive immediately
+ ipfix_decoder = IPFIXDecoder()
+ templates = ipfix.verify_templates(ipfix_decoder, count=1)
+
+ # make a tcp packet
+ self.pkts = [
+ (
+ Ether(src=self.pg3.remote_mac, dst=self.pg4.local_mac)
+ / IP(src=self.pg3.remote_ip4, dst=self.pg4.remote_ip4)
+ / TCP(sport=1234, dport=4321)
+ / Raw(b"\xa5" * 50)
+ )
+ ]
+
+ # send the tcp packet two times, each time with new set of flags
+ tcp_flags = (
+ TCP_F_SYN | TCP_F_ACK,
+ TCP_F_RST | TCP_F_PSH,
+ )
+ for f in tcp_flags:
+ self.pkts[0][TCP].flags = f
+ capture = self.send_packets(src_if=self.pg3, dst_if=self.pg4)
+
+ # verify meta info - packet/octet delta and tcp flags
+ cflow = self.wait_for_cflow_packet(self.collector, templates[0], timeout=6)
+ self.verify_cflow_data(ipfix_decoder, capture, cflow)
+ self.verify_cflow_data_detail(
+ ipfix_decoder,
+ capture,
+ cflow,
+ {
+ IPFIX_TCP_FLAGS_ID: f,
+ IPFIX_SRC_TRANS_PORT_ID: 1234,
+ IPFIX_DST_TRANS_PORT_ID: 4321,
+ },
+ )
+
+ self.collector.get_capture(3)
+
+ # cleanup
+ ipfix.remove_vpp_config()
+
def test_interface_dump(self):
"""Dump interfaces with IPFIX flow record generation enabled"""
self.logger.info("FFP_TEST_START_0003")
ipfix.remove_vpp_config()
self.logger.info("FFP_TEST_FINISH_0002")
+ def test_L234onL2(self):
+ """L2/3/4 data on L2 datapath"""
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pkts = []
+
+ ipfix = VppCFLOW(
+ test=self, intf=self.intf1, layer="l2 l3 l4", direction=self.direction
+ )
+ ipfix.add_vpp_config()
+
+ ipfix_decoder = IPFIXDecoder()
+ # template packet should arrive immediately
+ tmpl_l2_field_count = TMPL_COMMON_FIELD_COUNT + TMPL_L2_FIELD_COUNT
+ tmpl_ip_field_count = (
+ TMPL_COMMON_FIELD_COUNT
+ + TMPL_L2_FIELD_COUNT
+ + TMPL_L3_FIELD_COUNT
+ + TMPL_L4_FIELD_COUNT
+ )
+ templates = ipfix.verify_templates(
+ ipfix_decoder,
+ count=3,
+ field_count_in=(tmpl_l2_field_count, tmpl_ip_field_count),
+ )
+
+ # verify IPv4 and IPv6 flows
+ for ip_ver in ("v4", "v6"):
+ self.create_stream(packets=1, ip_ver=ip_ver)
+ capture = self.send_packets()
+
+ # make sure the one packet we expect actually showed up
+ self.vapi.ipfix_flush()
+ cflow = self.wait_for_cflow_packet(
+ self.collector, templates[1 if ip_ver == "v4" else 2]
+ )
+ src_ip_id = 8 if ip_ver == "v4" else 27
+ dst_ip_id = 12 if ip_ver == "v4" else 28
+ self.verify_cflow_data_detail(
+ ipfix_decoder,
+ capture,
+ cflow,
+ {
+ 2: "packets",
+ 256: 8 if ip_ver == "v4" else 56710,
+ 4: 17,
+ 7: "sport",
+ 11: "dport",
+ src_ip_id: "src_ip",
+ dst_ip_id: "dst_ip",
+ 61: (self.direction == "tx"),
+ },
+ ip_ver=ip_ver,
+ field_count=tmpl_ip_field_count,
+ )
+
+ # verify non-IP flow
+ self.pkts = [
+ (
+ Ether(dst=self.pg2.local_mac, src=self.pg1.remote_mac)
+ / SlowProtocol()
+ / LACP()
+ )
+ ]
+ capture = self.send_packets()
+
+ # make sure the one packet we expect actually showed up
+ self.vapi.ipfix_flush()
+ cflow = self.wait_for_cflow_packet(self.collector, templates[0])
+ self.verify_cflow_data_detail(
+ ipfix_decoder,
+ capture,
+ cflow,
+ {2: "packets", 256: 2440, 61: (self.direction == "tx")},
+ field_count=tmpl_l2_field_count,
+ )
+
+ self.collector.get_capture(6)
+
+ ipfix.remove_vpp_config()
+
def test_L4onL2(self):
"""L4 data on L2 datapath"""
self.logger.info("FFP_TEST_START_0003")
ipfix.remove_vpp_config()
self.logger.info("FFP_TEST_FINISH_0001")
+ def test_no_leftover_flows_after_disabling(self):
+ """disable flowprobe feature and expect no leftover flows"""
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pkts = []
+
+ # enable ip4 datapath for an interface
+ # set active and passive timers
+ ipfix = VppCFLOW(
+ test=self,
+ active=3,
+ passive=4,
+ intf="pg3",
+ layer="l3",
+ datapath="ip4",
+ direction="rx",
+ mtu=100,
+ )
+ ipfix.add_vpp_config()
+
+ # template packet should arrive immediately
+ ipfix.verify_templates(count=1)
+
+ # send some ip4 packets
+ self.create_stream(src_if=self.pg3, dst_if=self.pg4, packets=5)
+ self.send_packets(src_if=self.pg3, dst_if=self.pg4)
+
+ # disable feature for the interface
+ # currently stored ip4 flows should be removed
+ ipfix.disable_flowprobe_feature()
+
+ # no leftover ip4 flows are expected
+ self.pg_enable_capture([self.collector])
+ self.sleep(12, "wait for leftover ip4 flows during three passive intervals")
+ self.collector.assert_nothing_captured()
+
+ # cleanup
+ ipfix.disable_exporter()
+
@unittest.skipUnless(config.extended, "part of extended tests")
class ReenableFP(MethodHolder):
Code Review / vpp.git / blobdiff