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
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
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
}
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()
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]
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")
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)
Code Review / vpp.git / blobdiff