ligang/vpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2e7c5d33cc9dd7a79a0420d60b55169b620cc0c3
vpp/test/test_ipsec_spd_flow_cache.py
Zachary Leaf 26fec718f2 ipsec: input: drop by default for non-matching pkts 
As per IPSec RFC4301 [1], any non-matching packets should be dropped by
default. This is handled correctly in ipsec_output.c, however in
ipsec_input.c non-matching packets are allowed to pass as per a matched
BYPASS rule.

For full details, see:
https://lists.fd.io/g/vpp-dev/topic/ipsec_input_output_default/84943480

It appears the ipsec6_input_node only matches PROTECT policies. Until
this is extended to handle BYPASS + DISCARD, we may wish to not drop
by default here, since all IPv6 traffic not matching a PROTECT policy
will be dropped.

[1]: https://datatracker.ietf.org/doc/html/rfc4301

Type: fix
Signed-off-by: Zachary Leaf <zachary.leaf@arm.com>
Change-Id: Iddbfd008dbe082486d1928f6a10ffbd83d859a20
2022-03-08 17:43:43 +00:00

605 lines
27 KiB
Python
Raw Blame History

import socket
import unittest
from util import ppp
from framework import VppTestRunner
from template_ipsec import SpdFlowCacheTemplate
class SpdFlowCacheOutbound(SpdFlowCacheTemplate):
# Override setUpConstants to enable outbound flow cache in config
@classmethod
def setUpConstants(cls):
super(SpdFlowCacheOutbound, cls).setUpConstants()
cls.vpp_cmdline.extend(["ipsec", "{",
"ipv4-outbound-spd-flow-cache on",
"}"])
cls.logger.info("VPP modified cmdline is %s" % " "
.join(cls.vpp_cmdline))
class IPSec4SpdTestCaseAdd(SpdFlowCacheOutbound):
""" IPSec/IPv4 outbound: Policy mode test case with flow cache \
(add rule)"""
def test_ipsec_spd_outbound_add(self):
# In this test case, packets in IPv4 FWD path are configured
# to go through IPSec outbound SPD policy lookup.
# 2 SPD rules (1 HIGH and 1 LOW) are added.
# High priority rule action is set to BYPASS.
# Low priority rule action is set to DISCARD.
# Traffic sent on pg0 interface should match high priority
# rule and should be sent out on pg1 interface.
self.create_interfaces(2)
pkt_count = 5
self.spd_create_and_intf_add(1, [self.pg1])
policy_0 = self.spd_add_rem_policy( # outbound, priority 10
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
policy_1 = self.spd_add_rem_policy( # outbound, priority 5
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=5, policy_type="discard")
# check flow cache is empty before sending traffic
self.verify_num_outbound_flow_cache_entries(0)
# create the packet stream
packets = self.create_stream(self.pg0, self.pg1, pkt_count)
# add the stream to the source interface + enable capture
self.pg0.add_stream(packets)
self.pg0.enable_capture()
self.pg1.enable_capture()
# start the packet generator
self.pg_start()
# get capture
capture = self.pg1.get_capture()
for packet in capture:
try:
self.logger.debug(ppp("SPD - Got packet:", packet))
except Exception:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
self.logger.debug("SPD: Num packets: %s", len(capture.res))
# assert nothing captured on pg0
self.pg0.assert_nothing_captured()
# verify captured packets
self.verify_capture(self.pg0, self.pg1, capture)
# verify all policies matched the expected number of times
self.verify_policy_match(pkt_count, policy_0)
self.verify_policy_match(0, policy_1)
# check policy in SPD has been cached after traffic
# matched BYPASS rule in SPD
self.verify_num_outbound_flow_cache_entries(1)
class IPSec4SpdTestCaseRemove(SpdFlowCacheOutbound):
""" IPSec/IPv4 outbound: Policy mode test case with flow cache \
(remove rule)"""
def test_ipsec_spd_outbound_remove(self):
# In this test case, packets in IPv4 FWD path are configured
# to go through IPSec outbound SPD policy lookup.
# 2 SPD rules (1 HIGH and 1 LOW) are added.
# High priority rule action is set to BYPASS.
# Low priority rule action is set to DISCARD.
# High priority rule is then removed.
# Traffic sent on pg0 interface should match low priority
# rule and should be discarded after SPD lookup.
self.create_interfaces(2)
pkt_count = 5
self.spd_create_and_intf_add(1, [self.pg1])
policy_0 = self.spd_add_rem_policy( # outbound, priority 10
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
policy_1 = self.spd_add_rem_policy( # outbound, priority 5
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=5, policy_type="discard")
# check flow cache is empty before sending traffic
self.verify_num_outbound_flow_cache_entries(0)
# create the packet stream
packets = self.create_stream(self.pg0, self.pg1, pkt_count)
# add the stream to the source interface + enable capture
self.pg0.add_stream(packets)
self.pg0.enable_capture()
self.pg1.enable_capture()
# start the packet generator
self.pg_start()
# get capture
capture = self.pg1.get_capture()
for packet in capture:
try:
self.logger.debug(ppp("SPD - Got packet:", packet))
except Exception:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
# assert nothing captured on pg0
self.pg0.assert_nothing_captured()
# verify capture on pg1
self.logger.debug("SPD: Num packets: %s", len(capture.res))
self.verify_capture(self.pg0, self.pg1, capture)
# verify all policies matched the expected number of times
self.verify_policy_match(pkt_count, policy_0)
self.verify_policy_match(0, policy_1)
# check policy in SPD has been cached after traffic
# matched BYPASS rule in SPD
self.verify_num_outbound_flow_cache_entries(1)
# now remove the bypass rule
self.spd_add_rem_policy( # outbound, priority 10
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass",
remove=True)
# verify flow cache counter has been reset by rule removal
self.verify_num_outbound_flow_cache_entries(0)
# resend the same packets
self.pg0.add_stream(packets)
self.pg0.enable_capture() # flush the old captures
self.pg1.enable_capture()
self.pg_start()
# assert nothing captured on pg0
self.pg0.assert_nothing_captured()
# all packets will be dropped by SPD rule
self.pg1.assert_nothing_captured()
# verify all policies matched the expected number of times
self.verify_policy_match(pkt_count, policy_0)
self.verify_policy_match(pkt_count, policy_1)
# previous stale entry in flow cache should have been overwritten,
# with one active entry
self.verify_num_outbound_flow_cache_entries(1)
class IPSec4SpdTestCaseReadd(SpdFlowCacheOutbound):
""" IPSec/IPv4 outbound: Policy mode test case with flow cache \
(add, remove, re-add)"""
def test_ipsec_spd_outbound_readd(self):
# In this test case, packets in IPv4 FWD path are configured
# to go through IPSec outbound SPD policy lookup.
# 2 SPD rules (1 HIGH and 1 LOW) are added.
# High priority rule action is set to BYPASS.
# Low priority rule action is set to DISCARD.
# Traffic sent on pg0 interface should match high priority
# rule and should be sent out on pg1 interface.
# High priority rule is then removed.
# Traffic sent on pg0 interface should match low priority
# rule and should be discarded after SPD lookup.
# Readd high priority rule.
# Traffic sent on pg0 interface should match high priority
# rule and should be sent out on pg1 interface.
self.create_interfaces(2)
pkt_count = 5
self.spd_create_and_intf_add(1, [self.pg1])
policy_0 = self.spd_add_rem_policy( # outbound, priority 10
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
policy_1 = self.spd_add_rem_policy( # outbound, priority 5
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=5, policy_type="discard")
# check flow cache is empty before sending traffic
self.verify_num_outbound_flow_cache_entries(0)
# create the packet stream
packets = self.create_stream(self.pg0, self.pg1, pkt_count)
# add the stream to the source interface + enable capture
self.pg0.add_stream(packets)
self.pg0.enable_capture()
self.pg1.enable_capture()
# start the packet generator
self.pg_start()
# get capture
capture = self.pg1.get_capture()
for packet in capture:
try:
self.logger.debug(ppp("SPD - Got packet:", packet))
except Exception:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
self.logger.debug("SPD: Num packets: %s", len(capture.res))
# assert nothing captured on pg0
self.pg0.assert_nothing_captured()
# verify capture on pg1
self.verify_capture(self.pg0, self.pg1, capture)
# verify all policies matched the expected number of times
self.verify_policy_match(pkt_count, policy_0)
self.verify_policy_match(0, policy_1)
# check policy in SPD has been cached after traffic
# matched BYPASS rule in SPD
self.verify_num_outbound_flow_cache_entries(1)
# now remove the bypass rule, leaving only the discard rule
self.spd_add_rem_policy( # outbound, priority 10
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass",
remove=True)
# verify flow cache counter has been reset by rule removal
self.verify_num_outbound_flow_cache_entries(0)
# resend the same packets
self.pg0.add_stream(packets)
self.pg0.enable_capture() # flush the old captures
self.pg1.enable_capture()
self.pg_start()
# assert nothing captured on pg0
self.pg0.assert_nothing_captured()
# all packets will be dropped by SPD rule
self.pg1.assert_nothing_captured()
# verify all policies matched the expected number of times
self.verify_policy_match(pkt_count, policy_0)
self.verify_policy_match(pkt_count, policy_1)
# previous stale entry in flow cache should have been overwritten
self.verify_num_outbound_flow_cache_entries(1)
# now readd the bypass rule
policy_0 = self.spd_add_rem_policy( # outbound, priority 10
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
# verify flow cache counter has been reset by rule addition
self.verify_num_outbound_flow_cache_entries(0)
# resend the same packets
self.pg0.add_stream(packets)
self.pg0.enable_capture() # flush the old captures
self.pg1.enable_capture()
self.pg_start()
# get capture
capture = self.pg1.get_capture(pkt_count)
for packet in capture:
try:
self.logger.debug(ppp("SPD - Got packet:", packet))
except Exception:
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
self.logger.debug("SPD: Num packets: %s", len(capture.res))
# assert nothing captured on pg0
self.pg0.assert_nothing_captured()
# verify captured packets
self.verify_capture(self.pg0, self.pg1, capture)
# verify all policies matched the expected number of times
self.verify_policy_match(pkt_count, policy_0)
self.verify_policy_match(pkt_count, policy_1)
# previous stale entry in flow cache should have been overwritten
self.verify_num_outbound_flow_cache_entries(1)
class IPSec4SpdTestCaseMultiple(SpdFlowCacheOutbound):
""" IPSec/IPv4 outbound: Policy mode test case with flow cache \
(multiple interfaces, multiple rules)"""
def test_ipsec_spd_outbound_multiple(self):
# In this test case, packets in IPv4 FWD path are configured to go
# through IPSec outbound SPD policy lookup.
# Multiples rules on multiple interfaces are tested at the same time.
# 3x interfaces are configured, binding the same SPD to each.
# Each interface has 2 SPD rules (1 BYPASS and 1 DISCARD).
# On pg0 & pg1, the BYPASS rule is HIGH priority
# On pg2, the DISCARD rule is HIGH priority
# Traffic should be received on pg0 & pg1 and dropped on pg2.
self.create_interfaces(3)
pkt_count = 5
# bind SPD to all interfaces
self.spd_create_and_intf_add(1, self.pg_interfaces)
# add rules on all interfaces
policy_01 = self.spd_add_rem_policy( # outbound, priority 10
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
policy_02 = self.spd_add_rem_policy( # outbound, priority 5
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=5, policy_type="discard")
policy_11 = self.spd_add_rem_policy( # outbound, priority 10
1, self.pg1, self.pg2, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
policy_12 = self.spd_add_rem_policy( # outbound, priority 5
1, self.pg1, self.pg2, socket.IPPROTO_UDP,
is_out=1, priority=5, policy_type="discard")
policy_21 = self.spd_add_rem_policy( # outbound, priority 5
1, self.pg2, self.pg0, socket.IPPROTO_UDP,
is_out=1, priority=5, policy_type="bypass")
policy_22 = self.spd_add_rem_policy( # outbound, priority 10
1, self.pg2, self.pg0, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="discard")
# interfaces bound to an SPD, will by default drop inbound
# traffic with no matching policies. add catch-all inbound
# bypass rule to SPD:
self.spd_add_rem_policy( # inbound, all interfaces
1, None, None, socket.IPPROTO_UDP, is_out=0, priority=10,
policy_type="bypass", all_ips=True)
# check flow cache is empty (0 active elements) before sending traffic
self.verify_num_outbound_flow_cache_entries(0)
# create the packet streams
packets0 = self.create_stream(self.pg0, self.pg1, pkt_count)
packets1 = self.create_stream(self.pg1, self.pg2, pkt_count)
packets2 = self.create_stream(self.pg2, self.pg0, pkt_count)
# add the streams to the source interfaces
self.pg0.add_stream(packets0)
self.pg1.add_stream(packets1)
self.pg2.add_stream(packets2)
# enable capture on all interfaces
for pg in self.pg_interfaces:
pg.enable_capture()
# start the packet generator
self.pg_start()
# get captures
if_caps = []
for pg in [self.pg1, self.pg2]: # we are expecting captures on pg1/pg2
if_caps.append(pg.get_capture())
for packet in if_caps[-1]:
try:
self.logger.debug(ppp("SPD - Got packet:", packet))
except Exception:
self.logger.error(
ppp("Unexpected or invalid packet:", packet))
raise
self.logger.debug("SPD: Num packets: %s", len(if_caps[0].res))
self.logger.debug("SPD: Num packets: %s", len(if_caps[1].res))
# verify captures that matched BYPASS rule
self.verify_capture(self.pg0, self.pg1, if_caps[0])
self.verify_capture(self.pg1, self.pg2, if_caps[1])
# verify that traffic to pg0 matched DISCARD rule and was dropped
self.pg0.assert_nothing_captured()
# verify all packets that were expected to match rules, matched
# pg0 -> pg1
self.verify_policy_match(pkt_count, policy_01)
self.verify_policy_match(0, policy_02)
# pg1 -> pg2
self.verify_policy_match(pkt_count, policy_11)
self.verify_policy_match(0, policy_12)
# pg2 -> pg0
self.verify_policy_match(0, policy_21)
self.verify_policy_match(pkt_count, policy_22)
# check that 3 matching policies in SPD have been cached
self.verify_num_outbound_flow_cache_entries(3)
class IPSec4SpdTestCaseOverwriteStale(SpdFlowCacheOutbound):
""" IPSec/IPv4 outbound: Policy mode test case with flow cache \
(overwrite stale entries)"""
def test_ipsec_spd_outbound_overwrite(self):
# The operation of the flow cache is setup so that the entire cache
# is invalidated when adding or removing an SPD policy rule.
# For performance, old cache entries are not zero'd, but remain
# in the table as "stale" entries. If a flow matches a stale entry,
# and the epoch count does NOT match the current count, the entry
# is overwritten.
# In this test, 3 active rules are created and matched to enter
# them into the flow cache.
# A single entry is removed to invalidate the entire cache.
# We then readd the rule and test that overwriting of the previous
# stale entries occurs as expected, and that the flow cache entry
# counter is updated correctly.
self.create_interfaces(3)
pkt_count = 2
# bind SPD to all interfaces
self.spd_create_and_intf_add(1, self.pg_interfaces)
# add output rules on all interfaces
# pg0 -> pg1
policy_0 = self.spd_add_rem_policy( # outbound
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
# pg1 -> pg2
policy_1 = self.spd_add_rem_policy( # outbound
1, self.pg1, self.pg2, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
# pg2 -> pg0
policy_2 = self.spd_add_rem_policy( # outbound
1, self.pg2, self.pg0, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="discard")
# interfaces bound to an SPD, will by default drop inbound
# traffic with no matching policies. add catch-all inbound
# bypass rule to SPD:
self.spd_add_rem_policy( # inbound, all interfaces
1, None, None, socket.IPPROTO_UDP, is_out=0, priority=10,
policy_type="bypass", all_ips=True)
# check flow cache is empty (0 active elements) before sending traffic
self.verify_num_outbound_flow_cache_entries(0)
# create the packet streams
packets0 = self.create_stream(self.pg0, self.pg1, pkt_count)
packets1 = self.create_stream(self.pg1, self.pg2, pkt_count)
packets2 = self.create_stream(self.pg2, self.pg0, pkt_count)
# add the streams to the source interfaces
self.pg0.add_stream(packets0)
self.pg1.add_stream(packets1)
self.pg2.add_stream(packets2)
# enable capture on all interfaces
for pg in self.pg_interfaces:
pg.enable_capture()
# start the packet generator
self.pg_start()
# get captures from ifs
if_caps = []
for pg in [self.pg1, self.pg2]: # we are expecting captures on pg1/pg2
if_caps.append(pg.get_capture())
for packet in if_caps[-1]:
try:
self.logger.debug(ppp("SPD Add - Got packet:", packet))
except Exception:
self.logger.error(
ppp("Unexpected or invalid packet:", packet))
raise
# verify captures that matched BYPASS rules
self.verify_capture(self.pg0, self.pg1, if_caps[0])
self.verify_capture(self.pg1, self.pg2, if_caps[1])
# verify that traffic to pg0 matched DISCARD rule and was dropped
self.pg0.assert_nothing_captured()
# verify all policies matched the expected number of times
self.verify_policy_match(pkt_count, policy_0)
self.verify_policy_match(pkt_count, policy_1)
self.verify_policy_match(pkt_count, policy_2)
# check flow/policy match was cached for: 3x output policies
self.verify_num_outbound_flow_cache_entries(3)
# adding an inbound policy should not invalidate output flow cache
self.spd_add_rem_policy( # inbound
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=0, priority=10, policy_type="bypass")
# check flow cache counter has not been reset
self.verify_num_outbound_flow_cache_entries(3)
# remove a bypass policy - flow cache counter will be reset, and
# there will be 3x stale entries in flow cache
self.spd_add_rem_policy( # outbound
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass",
remove=True)
# readd policy
policy_0 = self.spd_add_rem_policy( # outbound
1, self.pg0, self.pg1, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
# check counter was reset with flow cache invalidation
self.verify_num_outbound_flow_cache_entries(0)
# resend the same packets
self.pg0.add_stream(packets0)
self.pg1.add_stream(packets1)
self.pg2.add_stream(packets2)
for pg in self.pg_interfaces:
pg.enable_capture() # flush previous captures
self.pg_start()
# get captures from ifs
if_caps = []
for pg in [self.pg1, self.pg2]: # we are expecting captures on pg1/pg2
if_caps.append(pg.get_capture())
for packet in if_caps[-1]:
try:
self.logger.debug(ppp("SPD Add - Got packet:", packet))
except Exception:
self.logger.error(
ppp("Unexpected or invalid packet:", packet))
raise
# verify captures that matched BYPASS rules
self.verify_capture(self.pg0, self.pg1, if_caps[0])
self.verify_capture(self.pg1, self.pg2, if_caps[1])
# verify that traffic to pg0 matched DISCARD rule and was dropped
self.pg0.assert_nothing_captured()
# verify all policies matched the expected number of times
self.verify_policy_match(pkt_count, policy_0)
self.verify_policy_match(pkt_count*2, policy_1)
self.verify_policy_match(pkt_count*2, policy_2)
# we are overwriting 3x stale entries - check flow cache counter
# is correct
self.verify_num_outbound_flow_cache_entries(3)
class IPSec4SpdTestCaseCollision(SpdFlowCacheOutbound):
""" IPSec/IPv4 outbound: Policy mode test case with flow cache \
(hash collision)"""
# Override class setup to restrict vector size to 16 elements.
# This forces using only the lower 4 bits of the hash as a key,
# making hash collisions easy to find.
@classmethod
def setUpConstants(cls):
super(SpdFlowCacheOutbound, cls).setUpConstants()
cls.vpp_cmdline.extend(["ipsec", "{",
"ipv4-outbound-spd-flow-cache on",
"ipv4-outbound-spd-hash-buckets 16",
"}"])
cls.logger.info("VPP modified cmdline is %s" % " "
.join(cls.vpp_cmdline))
def test_ipsec_spd_outbound_collision(self):
# The flow cache operation is setup to overwrite an entry
# if a hash collision occurs.
# In this test, 2 packets are configured that result in a
# hash with the same lower 4 bits.
# After the first packet is received, there should be one
# active entry in the flow cache.
# After the second packet with the same lower 4 bit hash
# is received, this should overwrite the same entry.
# Therefore there will still be a total of one (1) entry,
# in the flow cache with two matching policies.
# crc32_supported() method is used to check cpu for crc32
# intrinsic support for hashing.
# If crc32 is not supported, we fall back to clib_xxhash()
self.create_interfaces(3)
pkt_count = 5
# bind SPD to all interfaces
self.spd_create_and_intf_add(1, self.pg_interfaces)
# add rules
policy_0 = self.spd_add_rem_policy( # outbound, priority 10
1, self.pg1, self.pg2, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
policy_1 = self.spd_add_rem_policy( # outbound, priority 10
1, self.pg2, self.pg0, socket.IPPROTO_UDP,
is_out=1, priority=10, policy_type="bypass")
# interfaces bound to an SPD, will by default drop inbound
# traffic with no matching policies. add catch-all inbound
# bypass rule to SPD:
self.spd_add_rem_policy( # inbound, all interfaces
1, None, None, socket.IPPROTO_UDP, is_out=0, priority=10,
policy_type="bypass", all_ips=True)
# check flow cache is empty (0 active elements) before sending traffic
self.verify_num_outbound_flow_cache_entries(0)
# create the packet streams generating collision on last 4 bits
if self.crc32_supported():
# packet hashes to:
# 432c99c2
packets1 = self.create_stream(self.pg1, self.pg2, pkt_count, 1, 1)
# 31f8f3f2
packets2 = self.create_stream(self.pg2, self.pg0, pkt_count, 6, 6)
else: # clib_xxhash
# ec3a258551bc0306
packets1 = self.create_stream(self.pg1, self.pg2, pkt_count, 2, 2)
# 61fee526d18d7a6
packets2 = self.create_stream(self.pg2, self.pg0, pkt_count, 3, 3)
# add the streams to the source interfaces
self.pg1.add_stream(packets1)
self.pg2.add_stream(packets2)
# enable capture on all interfaces
for pg in self.pg_interfaces:
pg.enable_capture()
# start the packet generator
self.pg_start()
# get captures from ifs - the proper pkt_count of packets was saved by
# create_packet_info() based on dst_if parameter
if_caps = []
for pg in [self.pg2, self.pg0]: # we are expecting captures on pg2/pg0
if_caps.append(pg.get_capture())
for packet in if_caps[-1]:
try:
self.logger.debug(ppp(
"SPD - Got packet:", packet))
except Exception:
self.logger.error(ppp(
"Unexpected or invalid packet:", packet))
raise
self.logger.debug("SPD: Num packets: %s", len(if_caps[0].res))
self.logger.debug("SPD: Num packets: %s", len(if_caps[1].res))
# verify captures that matched BYPASS rule
self.verify_capture(self.pg1, self.pg2, if_caps[0])
self.verify_capture(self.pg2, self.pg0, if_caps[1])
# verify all packets that were expected to match rules, matched
self.verify_policy_match(pkt_count, policy_0)
self.verify_policy_match(pkt_count, policy_1)
# we have matched 2 policies, but due to the hash collision
# one active entry is expected
self.verify_num_outbound_flow_cache_entries(1)
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
Reference in New Issue View Git Blame Copy Permalink