diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/test_acl_plugin_conns.py | 150 |
1 files changed, 140 insertions, 10 deletions
diff --git a/test/test_acl_plugin_conns.py b/test/test_acl_plugin_conns.py index 1a9100cb..06f3cf7e 100644 --- a/test/test_acl_plugin_conns.py +++ b/test/test_acl_plugin_conns.py @@ -5,7 +5,7 @@ import unittest from framework import VppTestCase, VppTestRunner, running_extended_tests from scapy.layers.l2 import Ether from scapy.packet import Raw -from scapy.layers.inet import IP, UDP +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 @@ -81,7 +81,7 @@ class Conn(): self.ports[1] = port2 self - def pkt(self, side): + def pkt(self, side, flags=None): is_ip6 = 1 if self.address_family == AF_INET6 else 0 s0 = side s1 = 1-side @@ -90,9 +90,12 @@ class Conn(): layer_3 = [IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4), IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)] payload = "x" + l4args = {'sport': self.ports[s0], 'dport': self.ports[s1]} + if flags is not None: + l4args['flags'] = flags p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) / layer_3[is_ip6] / - self.l4proto(sport=self.ports[s0], dport=self.ports[s1]) / + self.l4proto(**l4args) / Raw(payload)) return p @@ -149,8 +152,8 @@ class Conn(): } return new_rule - def send(self, side): - self.ifs[side].add_stream(self.pkt(side)) + def send(self, side, flags=None): + self.ifs[side].add_stream(self.pkt(side, flags)) self.ifs[1-side].enable_capture() self.testcase.pg_start() @@ -158,14 +161,14 @@ class Conn(): p = self.ifs[side].wait_for_packet(1) return p - def send_through(self, side): - self.send(side) + def send_through(self, side, flags=None): + self.send(side, flags) p = self.recv(1-side) return p - def send_pingpong(self, side): - p1 = self.send_through(side) - p2 = self.send_through(1-side) + def send_pingpong(self, side, flags1=None, flags2=None): + p1 = self.send_through(side, flags1) + p2 = self.send_through(1-side, flags2) return [p1, p2] @@ -300,6 +303,77 @@ class ACLPluginConnTestCase(VppTestCase): p2 = None self.assert_equal(p2, None, "packet on supposedly deleted conn") + 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') + # the return packets should pass + conn1.send_through(1, 'SA') + # allow the conn to time out + for i in IterateWithSleep(self, 30, "Wait for timeout", 0.1): + pass + # ensure conn times out + try: + p2 = conn1.send_through(1).command() + except: + # If we asserted while waiting, it's good. + # the conn should have timed out. + p2 = None + self.assert_equal(p2, None, "packet on supposedly deleted conn") + + 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') + # the return packets should pass + conn1.send_through(1, 'SA') + # complete the threeway handshake + # (NB: sequence numbers not tracked, so not set!) + 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') + # ensure conn times out for real + for i in IterateWithSleep(self, 130, "Wait for timeout", 0.1): + pass + try: + p2 = conn1.send_through(1).command() + except: + # If we asserted while waiting, it's good. + # the conn should have timed out. + p2 = None + self.assert_equal(p2, None, "packet on supposedly deleted conn") + + 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') + # the return packets should pass + conn1.send_through(1, 'SA') + # complete the threeway handshake + # (NB: sequence numbers not tracked, so not set!) + 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') + # Send the FIN to bounce the session out of established + 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 + # Now it should have timed out already + try: + p2 = conn1.send_through(1).command() + except: + # If we asserted while waiting, it's good. + # the conn should have timed out. + p2 = None + self.assert_equal(p2, None, "packet on supposedly deleted conn") + def test_0000_conn_prepare_test(self): """ Prepare the settings """ self.vapi.ppcli("set acl-plugin session timeout udp idle 1") @@ -351,3 +425,59 @@ class ACLPluginConnTestCase(VppTestCase): def test_1012_active_conn_test(self): """ 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 """ + # 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 + self.vapi.ppcli("set acl-plugin session timeout tcp idle 10") + 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 """ + 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 """ + 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 """ + 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 """ + 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 """ + 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 """ + 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 """ + 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 """ + 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 """ + 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 """ + 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 """ + 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 """ + self.run_tcp_transient_teardown_conn_test(AF_INET6, 1) |