diff options
Diffstat (limited to 'test/test_acl_plugin_conns.py')
| -rw-r--r-- | test/test_acl_plugin_conns.py | 169 |
1 files changed, 87 insertions, 82 deletions
diff --git a/test/test_acl_plugin_conns.py b/test/test_acl_plugin_conns.py index c7941fa150b..3f0a6d9faf6 100644 --- a/test/test_acl_plugin_conns.py +++ b/test/test_acl_plugin_conns.py @@ -2,17 +2,12 @@ """ ACL plugin extended stateful tests """ import unittest -from framework import VppTestCase, VppTestRunner, running_extended_tests -from scapy.layers.l2 import Ether -from scapy.packet import Raw +from config import config +from framework import VppTestCase from scapy.layers.inet import IP, UDP, TCP from scapy.packet import Packet -from socket import inet_pton, AF_INET, AF_INET6 -from scapy.layers.inet6 import IPv6, ICMPv6Unknown, ICMPv6EchoRequest -from scapy.layers.inet6 import ICMPv6EchoReply, IPv6ExtHdrRouting -from scapy.layers.inet6 import IPv6ExtHdrFragment -from pprint import pprint -from random import randint +from socket import AF_INET, AF_INET6 +from scapy.layers.inet6 import IPv6 from util import L4_Conn from ipaddress import ip_network @@ -38,14 +33,16 @@ def to_acl_rule(self, is_permit, wildcard_sport=False): rule_l4_sport_first = rule_l4_sport rule_l4_sport_last = rule_l4_sport - new_rule = AclRule(is_permit=is_permit, proto=rule_l4_proto, - src_prefix=ip_network( - (p[rule_l3_layer].src, rule_prefix_len)), - dst_prefix=ip_network( - (p[rule_l3_layer].dst, rule_prefix_len)), - sport_from=rule_l4_sport_first, - sport_to=rule_l4_sport_last, - dport_from=rule_l4_dport, dport_to=rule_l4_dport) + new_rule = AclRule( + is_permit=is_permit, + proto=rule_l4_proto, + src_prefix=ip_network((p[rule_l3_layer].src, rule_prefix_len)), + dst_prefix=ip_network((p[rule_l3_layer].dst, rule_prefix_len)), + sport_from=rule_l4_sport_first, + sport_to=rule_l4_sport_last, + dport_from=rule_l4_dport, + dport_to=rule_l4_dport, + ) return new_rule @@ -53,7 +50,7 @@ def to_acl_rule(self, is_permit, wildcard_sport=False): Packet.to_acl_rule = to_acl_rule -class IterateWithSleep(): +class IterateWithSleep: def __init__(self, testcase, n_iters, description, sleep_sec): self.curr = 0 self.testcase = testcase @@ -86,21 +83,27 @@ class Conn(L4_Conn): deny_acl.add_vpp_config() if reflect_side == acl_side: - acl_if0 = VppAclInterface(self.testcase, - self.ifs[acl_side].sw_if_index, - [reflect_acl, deny_acl], n_input=1) - acl_if1 = VppAclInterface(self.testcase, - self.ifs[1-acl_side].sw_if_index, [], - n_input=0) + acl_if0 = VppAclInterface( + self.testcase, + self.ifs[acl_side].sw_if_index, + [reflect_acl, deny_acl], + n_input=1, + ) + acl_if1 = VppAclInterface( + self.testcase, self.ifs[1 - acl_side].sw_if_index, [], n_input=0 + ) acl_if0.add_vpp_config() acl_if1.add_vpp_config() else: - acl_if0 = VppAclInterface(self.testcase, - self.ifs[acl_side].sw_if_index, - [deny_acl, reflect_acl], n_input=1) - acl_if1 = VppAclInterface(self.testcase, - self.ifs[1-acl_side].sw_if_index, [], - n_input=0) + acl_if0 = VppAclInterface( + self.testcase, + self.ifs[acl_side].sw_if_index, + [deny_acl, reflect_acl], + n_input=1, + ) + acl_if1 = VppAclInterface( + self.testcase, self.ifs[1 - acl_side].sw_if_index, [], n_input=0 + ) acl_if0.add_vpp_config() acl_if1.add_vpp_config() @@ -108,19 +111,22 @@ class Conn(L4_Conn): any_addr = ["0.0.0.0", "::"] rule_family = self.address_family is_ip6 = 1 if rule_family == AF_INET6 else 0 - new_rule = AclRule(is_permit=is_permit, proto=0, - src_prefix=ip_network( - (any_addr[is_ip6], 0)), - dst_prefix=ip_network( - (any_addr[is_ip6], 0)), - sport_from=0, sport_to=65535, dport_from=0, - dport_to=65535) + new_rule = AclRule( + is_permit=is_permit, + proto=0, + src_prefix=ip_network((any_addr[is_ip6], 0)), + dst_prefix=ip_network((any_addr[is_ip6], 0)), + sport_from=0, + sport_to=65535, + dport_from=0, + dport_to=65535, + ) return new_rule -@unittest.skipUnless(running_extended_tests, "part of extended tests") +@unittest.skipUnless(config.extended, "part of extended tests") class ACLPluginConnTestCase(VppTestCase): - """ ACL plugin connection-oriented extended testcases """ + """ACL plugin connection-oriented extended testcases""" @classmethod def setUpClass(cls): @@ -141,8 +147,7 @@ class ACLPluginConnTestCase(VppTestCase): super(ACLPluginConnTestCase, cls).tearDownClass() def tearDown(self): - """Run standard test teardown and log various show commands - """ + """Run standard test teardown and log various show commands""" super(ACLPluginConnTestCase, self).tearDown() def show_commands_at_teardown(self): @@ -155,7 +160,7 @@ class ACLPluginConnTestCase(VppTestCase): self.logger.info(self.vapi.cli("show event-logger all")) def run_basic_conn_test(self, af, acl_side): - """ Basic conn timeout test """ + """Basic conn timeout test""" conn1 = Conn(self, self.pg0, self.pg1, af, UDP, 42001, 4242) conn1.apply_acls(0, acl_side) conn1.send_through(0) @@ -177,8 +182,8 @@ class ACLPluginConnTestCase(VppTestCase): self.assert_equal(p2, None, "packet on long-idle conn") def run_active_conn_test(self, af, acl_side): - """ Idle connection behind active connection test """ - base = 10000 + 1000*acl_side + """Idle connection behind active connection test""" + base = 10000 + 1000 * acl_side conn1 = Conn(self, self.pg0, self.pg1, af, UDP, base + 1, 2323) conn2 = Conn(self, self.pg0, self.pg1, af, UDP, base + 2, 2323) conn3 = Conn(self, self.pg0, self.pg1, af, UDP, base + 3, 2323) @@ -205,7 +210,7 @@ class ACLPluginConnTestCase(VppTestCase): self.assert_equal(p2, None, "packet on long-idle conn") def run_clear_conn_test(self, af, acl_side): - """ Clear the connections via CLI """ + """Clear the connections via CLI""" conn1 = Conn(self, self.pg0, self.pg1, af, UDP, 42001, 4242) conn1.apply_acls(0, acl_side) conn1.send_through(0) @@ -228,9 +233,9 @@ class ACLPluginConnTestCase(VppTestCase): def run_tcp_transient_setup_conn_test(self, af, acl_side): conn1 = Conn(self, self.pg0, self.pg1, af, TCP, 53001, 5151) conn1.apply_acls(0, acl_side) - conn1.send_through(0, 'S') + conn1.send_through(0, "S") # the return packets should pass - conn1.send_through(1, 'SA') + conn1.send_through(1, "SA") # allow the conn to time out for i in IterateWithSleep(self, 30, "Wait for timeout", 0.1): pass @@ -246,17 +251,17 @@ class ACLPluginConnTestCase(VppTestCase): def run_tcp_established_conn_test(self, af, acl_side): conn1 = Conn(self, self.pg0, self.pg1, af, TCP, 53002, 5052) conn1.apply_acls(0, acl_side) - conn1.send_through(0, 'S') + conn1.send_through(0, "S") # the return packets should pass - conn1.send_through(1, 'SA') + conn1.send_through(1, "SA") # complete the threeway handshake # (NB: sequence numbers not tracked, so not set!) - conn1.send_through(0, 'A') + conn1.send_through(0, "A") # allow the conn to time out if it's in embryonic timer for i in IterateWithSleep(self, 30, "Wait for transient timeout", 0.1): pass # Try to send the packet from the "forbidden" side - it must pass - conn1.send_through(1, 'A') + conn1.send_through(1, "A") # ensure conn times out for real for i in IterateWithSleep(self, 130, "Wait for timeout", 0.1): pass @@ -271,19 +276,19 @@ class ACLPluginConnTestCase(VppTestCase): def run_tcp_transient_teardown_conn_test(self, af, acl_side): conn1 = Conn(self, self.pg0, self.pg1, af, TCP, 53002, 5052) conn1.apply_acls(0, acl_side) - conn1.send_through(0, 'S') + conn1.send_through(0, "S") # the return packets should pass - conn1.send_through(1, 'SA') + conn1.send_through(1, "SA") # complete the threeway handshake # (NB: sequence numbers not tracked, so not set!) - conn1.send_through(0, 'A') + conn1.send_through(0, "A") # allow the conn to time out if it's in embryonic timer for i in IterateWithSleep(self, 30, "Wait for transient timeout", 0.1): pass # Try to send the packet from the "forbidden" side - it must pass - conn1.send_through(1, 'A') + conn1.send_through(1, "A") # Send the FIN to bounce the session out of established - conn1.send_through(1, 'FA') + conn1.send_through(1, "FA") # If conn landed on transient timer it will time out here for i in IterateWithSleep(self, 30, "Wait for transient timeout", 0.1): pass @@ -297,59 +302,59 @@ class ACLPluginConnTestCase(VppTestCase): self.assert_equal(p2, None, "packet on supposedly deleted conn") def test_0000_conn_prepare_test(self): - """ Prepare the settings """ + """Prepare the settings""" self.vapi.ppcli("set acl-plugin session timeout udp idle 1") def test_0001_basic_conn_test(self): - """ IPv4: Basic conn timeout test reflect on ingress """ + """IPv4: Basic conn timeout test reflect on ingress""" self.run_basic_conn_test(AF_INET, 0) def test_0002_basic_conn_test(self): - """ IPv4: Basic conn timeout test reflect on egress """ + """IPv4: Basic conn timeout test reflect on egress""" self.run_basic_conn_test(AF_INET, 1) def test_0005_clear_conn_test(self): - """ IPv4: reflect egress, clear conn """ + """IPv4: reflect egress, clear conn""" self.run_clear_conn_test(AF_INET, 1) def test_0006_clear_conn_test(self): - """ IPv4: reflect ingress, clear conn """ + """IPv4: reflect ingress, clear conn""" self.run_clear_conn_test(AF_INET, 0) def test_0011_active_conn_test(self): - """ IPv4: Idle conn behind active conn, reflect on ingress """ + """IPv4: Idle conn behind active conn, reflect on ingress""" self.run_active_conn_test(AF_INET, 0) def test_0012_active_conn_test(self): - """ IPv4: Idle conn behind active conn, reflect on egress """ + """IPv4: Idle conn behind active conn, reflect on egress""" self.run_active_conn_test(AF_INET, 1) def test_1001_basic_conn_test(self): - """ IPv6: Basic conn timeout test reflect on ingress """ + """IPv6: Basic conn timeout test reflect on ingress""" self.run_basic_conn_test(AF_INET6, 0) def test_1002_basic_conn_test(self): - """ IPv6: Basic conn timeout test reflect on egress """ + """IPv6: Basic conn timeout test reflect on egress""" self.run_basic_conn_test(AF_INET6, 1) def test_1005_clear_conn_test(self): - """ IPv6: reflect egress, clear conn """ + """IPv6: reflect egress, clear conn""" self.run_clear_conn_test(AF_INET6, 1) def test_1006_clear_conn_test(self): - """ IPv6: reflect ingress, clear conn """ + """IPv6: reflect ingress, clear conn""" self.run_clear_conn_test(AF_INET6, 0) def test_1011_active_conn_test(self): - """ IPv6: Idle conn behind active conn, reflect on ingress """ + """IPv6: Idle conn behind active conn, reflect on ingress""" self.run_active_conn_test(AF_INET6, 0) def test_1012_active_conn_test(self): - """ IPv6: Idle conn behind active conn, reflect on egress """ + """IPv6: Idle conn behind active conn, reflect on egress""" self.run_active_conn_test(AF_INET6, 1) def test_2000_prepare_for_tcp_test(self): - """ Prepare for TCP session tests """ + """Prepare for TCP session tests""" # ensure the session hangs on if it gets treated as UDP self.vapi.ppcli("set acl-plugin session timeout udp idle 200") # let the TCP connection time out at 5 seconds @@ -357,49 +362,49 @@ class ACLPluginConnTestCase(VppTestCase): self.vapi.ppcli("set acl-plugin session timeout tcp transient 1") def test_2001_tcp_transient_conn_test(self): - """ IPv4: transient TCP session (incomplete 3WHS), ref. on ingress """ + """IPv4: transient TCP session (incomplete 3WHS), ref. on ingress""" self.run_tcp_transient_setup_conn_test(AF_INET, 0) def test_2002_tcp_transient_conn_test(self): - """ IPv4: transient TCP session (incomplete 3WHS), ref. on egress """ + """IPv4: transient TCP session (incomplete 3WHS), ref. on egress""" self.run_tcp_transient_setup_conn_test(AF_INET, 1) def test_2003_tcp_transient_conn_test(self): - """ IPv4: established TCP session (complete 3WHS), ref. on ingress """ + """IPv4: established TCP session (complete 3WHS), ref. on ingress""" self.run_tcp_established_conn_test(AF_INET, 0) def test_2004_tcp_transient_conn_test(self): - """ IPv4: established TCP session (complete 3WHS), ref. on egress """ + """IPv4: established TCP session (complete 3WHS), ref. on egress""" self.run_tcp_established_conn_test(AF_INET, 1) def test_2005_tcp_transient_teardown_conn_test(self): - """ IPv4: transient TCP session (3WHS,ACK,FINACK), ref. on ingress """ + """IPv4: transient TCP session (3WHS,ACK,FINACK), ref. on ingress""" self.run_tcp_transient_teardown_conn_test(AF_INET, 0) def test_2006_tcp_transient_teardown_conn_test(self): - """ IPv4: transient TCP session (3WHS,ACK,FINACK), ref. on egress """ + """IPv4: transient TCP session (3WHS,ACK,FINACK), ref. on egress""" self.run_tcp_transient_teardown_conn_test(AF_INET, 1) def test_3001_tcp_transient_conn_test(self): - """ IPv6: transient TCP session (incomplete 3WHS), ref. on ingress """ + """IPv6: transient TCP session (incomplete 3WHS), ref. on ingress""" self.run_tcp_transient_setup_conn_test(AF_INET6, 0) def test_3002_tcp_transient_conn_test(self): - """ IPv6: transient TCP session (incomplete 3WHS), ref. on egress """ + """IPv6: transient TCP session (incomplete 3WHS), ref. on egress""" self.run_tcp_transient_setup_conn_test(AF_INET6, 1) def test_3003_tcp_transient_conn_test(self): - """ IPv6: established TCP session (complete 3WHS), ref. on ingress """ + """IPv6: established TCP session (complete 3WHS), ref. on ingress""" self.run_tcp_established_conn_test(AF_INET6, 0) def test_3004_tcp_transient_conn_test(self): - """ IPv6: established TCP session (complete 3WHS), ref. on egress """ + """IPv6: established TCP session (complete 3WHS), ref. on egress""" self.run_tcp_established_conn_test(AF_INET6, 1) def test_3005_tcp_transient_teardown_conn_test(self): - """ IPv6: transient TCP session (3WHS,ACK,FINACK), ref. on ingress """ + """IPv6: transient TCP session (3WHS,ACK,FINACK), ref. on ingress""" self.run_tcp_transient_teardown_conn_test(AF_INET6, 0) def test_3006_tcp_transient_teardown_conn_test(self): - """ IPv6: transient TCP session (3WHS,ACK,FINACK), ref. on egress """ + """IPv6: transient TCP session (3WHS,ACK,FINACK), ref. on egress""" self.run_tcp_transient_teardown_conn_test(AF_INET6, 1) |