NAME = 'PyYAML'
VERSION = '3.01'
DESCRIPTION = "YAML parser and emitter for Python"
LONG_DESCRIPTION = """\
YAML is a data serialization format designed for human readability and
interaction with scripting languages. PyYAML is a YAML parser and
emitter for Python.
PyYAML features a complete YAML 1.1 parser, Unicode support, pickle
support, capable extension API, and sensible error messages. PyYAML
supports standard YAML tags and provides Python-specific tags that allow
to represent an arbitrary Python object.
PyYAML is applicable for a broad range of tasks from complex
configuration files to object serialization and persistance."""
AUTHOR = "Kirill Simonov"
AUTHOR_EMAIL = 'xi@resolvent.net'
LICENSE = "MIT"
PLATFORMS = "Any"
URL = "http://pyyaml.org/wiki/PyYAML"
DOWNLOAD_URL = "http://pyyaml.org/download/pyyaml/%s-%s.tar.gz" % (NAME, VERSION)
CLASSIFIERS = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: Text Processing :: Markup",
]
from distutils.core import setup
setup(
name=NAME,
version=VERSION,
description=DESCRIPTION,
long_description=LONG_DESCRIPTION,
author=AUTHOR,
author_email=AUTHOR_EMAIL,
license=LICENSE,
platforms=PLATFORMS,
url
@media only all and (prefers-color-scheme: dark) {
.highlight .hll { background-color: #49483e }
.highlight .c { color: #75715e } /* Comment */
.highlight .err { color: #960050; background-color: #1e0010 } /* Error */
.highlight .k { color: #66d9ef } /* Keyword */
.highlight .l { color: #ae81ff } /* Literal */
.highlight .n { color: #f8f8f2 } /* Name */
.highlight .o { color: #f92672 } /* Operator */
.highlight .p { color: #f8f8f2 } /* Punctuation */
.highlight .ch { color: #75715e } /* Comment.Hashbang */
.highlight .cm { color: #75715e } /* Comment.Multiline */
.highlight .cp { color: #75715e } /* Comment.Preproc */
.highlight .cpf { color: #75715e } /* Comment.PreprocFile */
.highlight .c1 { color: #75715e } /* Comment.Single */
.highlight .cs { color: #75715e } /* Comment.Special */
.highlight .gd { color: #f92672 } /* Generic.Deleted */
.highlight .ge { font-style: italic } /* Generic.Emph */
.highlight .gi { color: #a6e22e } /* Generic.Inserted */
.highlight .gs { font-weight: bold } /* Generic.Strong */
.highlight .gu { color: #75715e } /* Generic.Subheading */
.highlight .kc { color: #66d9ef } /* Keyword.Constant */
.highlight .kd { color: #66d9ef } /* Keyword.Declaration */
.highlight .kn { color: #f92672 } /* Keyword.Namespace */
.highlight .kp { color: #66d9ef } /* Keyword.Pseudo */
.highlight .kr { color: #66d9ef } /* Keyword.Reserved */
.highlight .kt { color: #66d9ef } /* Keyword.Type */
.highlight .ld { color: #e6db74 } /* Literal.Date */
.highlight .m { color: #ae81ff } /* Literal.Number */
.highlight .s { color: #e6db74 } /* Literal.String */
.highlight .na { color: #a6e22e } /* Name.Attribute */
.highlight .nb { color: #f8f8f2 } /* Name.Builtin */
.highlight .nc { color: #a6e22e } /* Name.Class */
.highlight .no { color: #66d9ef } /* Name.Constant */
.highlight .nd { color: #a6e22e } /* Name.Decorator */
.highlight .ni { color: #f8f8f2 } /* Name.Entity */
.highlight .ne { color: #a6e22e } /* Name.Exception */
.highlight .nf { color: #a6e22e } /* Name.Function */
.highlight .nl { color: #f8f8f2 } /* Name.Label */
.highlight .nn { color: #f8f8f2 } /* Name.Namespace */
.highlight .nx { color: #a6e22e } /* Name.Other */
.highlight .py { color: #f8f8f2 } /* Name.Property */
.highlight .nt { color: #f92672 } /* Name.Tag */
.highlight .nv { color: #f8f8f2 } /* Name.Variable */
.highlight .ow { color: #f92672 } /* Operator.Word */
.highlight .w { color: #f8f8f2 } /* Text.Whitespace */
.highlight .mb { color: #ae81ff } /* Literal.Number.Bin */
.highlight .mf { color: #ae81ff } /* Literal.Number.Float */
.highlight .mh { color: #ae81ff } /* Literal.Number.Hex */
.highlight .mi { color: #ae81ff } /* Literal.Number.Integer */
.highlight .mo { color: #ae81ff } /* Literal.Number.Oct */
.highlight .sa { color: #e6db74 } /* Literal.String.Affix */
.highlight .sb { color: #e6db74 } /* Literal.String.Backtick */
.highlight .sc { color: #e6db74 } /* Literal.String.Char */
.highlight .dl { color: #e6db74 } /* Literal.String.Delimiter */
.highlight .sd { color: #e6db74 } /* Literal.String.Doc */
.highlight .s2 { color: #e6db74 } /* Literal.String.Double */
.highlight .se { color: #ae81ff } /* Literal.String.Escape */
.highlight .sh { color: #e6db74 } /* Literal.String.Heredoc */
.highlight .si { color: #e6db74 } /* Literal.String.Interpol */
.highlight .sx { color: #e6db74 } /* Literal.String.Other */
.highlight .sr { color: #e6db74 } /* Literal.String.Regex */
.highlight .s1 { color: #e6db74 } /* Literal.String.Single */
.highlight .ss { color: #e6db74 } /* Literal.String.Symbol */
.highlight .bp { color: #f8f8f2 } /* Name.Builtin.Pseudo */
.highlight .fm { color: #a6e22e } /* Name.Function.Magic */
.highlight .vc { color: #f8f8f2 } /* Name.Variable.Class */
.highlight .vg { color: #f8f8f2 } /* Name.Variable.Global */
.highlight .vi { color: #f8f8f2 } /* Name.Variable.Instance */
.highlight .vm { color: #f8f8f2 } /* Name.Variable.Magic */
.highlight .il { color: #ae81ff } /* Literal.Number.Integer.Long */
}
@media (prefers-color-scheme: light) {
.highlight .hll { background-color: #ffffcc }
.highlight .c { color: #888888 } /* Comment */
.highlight .err { color: #a61717; background-color: #e3d2d2 } /* Error */
.highlight .k { color: #008800; font-weight: bold } /* Keyword */
.highlight .ch { color: #888888 } /* Comment.Hashbang */
.highlight .cm { color: #888888 } /* Comment.Multiline */
.highlight .cp { color: #cc0000; font-weight: bold } /* Comment.Preproc */
.highlight .cpf { color: #888888 } /* Comment.PreprocFile */
.highlight .c1 { color: #888888 } /* Comment.Single */
.highlight .cs { color: #cc0000; font-weight: bold; background-color: #fff0f0 } /* Comment.Special */
.highlight .gd { color: #000000; background-color: #ffdddd } /* Generic.Deleted */
.highlight .ge { font-style: italic } /* Generic.Emph */
.highlight .gr { color: #aa0000 } /* Generic.Error */
.highlight .gh { color: #333333 } /* Generic.Heading */
.highlight .gi { color: #000000; background-color: #ddffdd } /* Generic.Inserted */
.highlight .go { color: #888888 } /* Generic.Output */
.highlight .gp { color: #555555 } /* Generic.Prompt */
.highlight .gs { font-weight: bold } /* Generic.Strong */
.highlight .gu { color: #666666 } /* Generic.Subheading */
.highlight .gt { color: #aa0000 } /* Generic.Traceback */
.highlight .kc { color: #008800; font-weight: bold } /* Keyword.Constant */
.highlight .kd { color: #008800; font-weight: bold } /* Keyword.Declaration */
.highlight .kn { color: #008800; font-weight: bold } /* Keyword.Namespace */
.highlight .kp { color: #008800 } /* Keyword.Pseudo */
.highlight .kr { color: #008800; font-weight: bold } /* Keyword.Reserved */
.highlight .kt { color: #888888; font-weight: bold } /* Keyword.Type */
.highlight .m { color: #0000DD; font-weight: bold } /* Literal.Number */
.highlight .s { color: #dd2200; background-color: #fff0f0 } /* Literal.String */
.highlight .na { color: #336699 } /* Name.Attribute */
.highlight .nb { color: #003388 } /* Name.Builtin */
.highlight .nc { color: #bb0066; font-weight: bold } /* Name.Class */
.highlight .no { color: #003366; font-weight: bold } /* Name.Constant */
.highlight .nd { color: #555555 } /* Name.Decorator */
.highlight .ne { color: #bb0066; font-weight: bold } /* Name.Exception */
.highlight .nf { color: #0066bb; font-weight: bold } /* Name.Function */
.highlight .nl { color: #336699; font-style: italic } /* Name.Label */
.highlight .nn { color: #bb0066; font-weight: bold } /* Name.Namespace */
.highlight .py { color: #336699; font-weight: bold } /* Name.Property */
.highlight .nt { color: #bb0066; font-weight: bold } /* Name.Tag */
.highlight .nv { color: #336699 } /* Name.Variable */
.highlight .ow { color: #008800 } /* Operator.Word */
.highlight .w { color: #bbbbbb } /* Text.Whitespace */
.highlight .mb { color: #0000DD; font-weight: bold } /* Literal.Number.Bin */
.highlight .mf { color: #0000DD; font-weight: bold } /* Literal.Number.Float */
.highlight .mh { color: #0000DD; font-weight: bold } /* Literal.Number.Hex */
.highlight .mi { color: #0000DD; font-weight: bold } /* Literal.Number.Integer */
.highlight .mo { color: #0000DD; font-weight: bold } /* Literal.Number.Oct */
.highlight .sa { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Affix */
.highlight .sb { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Backtick */
.highlight .sc { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Char */
.highlight .dl { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Delimiter */
.highlight .sd { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Doc */
.highlight .s2 { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Double */
.highlight .se { color: #0044dd; background-color: #fff0f0 } /* Literal.String.Escape */
.highlight .sh { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Heredoc */
.highlight .si { color: #3333bb; background-color: #fff0f0 } /* Literal.String.Interpol */
.highlight .sx { color: #22bb22; background-color: #f0fff0 } /* Literal.String.Other */
.highlight .sr { color: #008800; background-color: #fff0ff } /* Literal.String.Regex */
.highlight .s1 { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Single */
.highlight .ss { color: #aa6600; background-color: #fff0f0 } /* Literal.String.Symbol */
.highlight .bp { color: #003388 } /* Name.Builtin.Pseudo */
.highlight .fm { color: #0066bb; font-weight: bold } /* Name.Function.Magic */
.highlight .vc { color: #336699 } /* Name.Variable.Class */
.highlight .vg { color: #dd7700 } /* Name.Variable.Global */
.highlight .vi { color: #3333bb } /* Name.Variable.Instance */
.highlight .vm { color: #336699 } /* Name.Variable.Magic */
.highlight .il { color: #0000DD; font-weight: bold } /* Literal.Number.Integer.Long */
}
#!/usr/bin/env python
""" 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 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 util import L4_Conn
def to_acl_rule(self, is_permit, wildcard_sport=False):
p = self
rule_family = AF_INET6 if p.haslayer(IPv6) else AF_INET
rule_prefix_len = 128 if p.haslayer(IPv6) else 32
rule_l3_layer = IPv6 if p.haslayer(IPv6) else IP
rule_l4_sport = p.sport
rule_l4_dport = p.dport
if p.haslayer(IPv6):
rule_l4_proto = p[IPv6].nh
else:
rule_l4_proto = p[IP].proto
if wildcard_sport:
rule_l4_sport_first = 0
rule_l4_sport_last = 65535
else:
rule_l4_sport_first = rule_l4_sport
rule_l4_sport_last = rule_l4_sport
new_rule = {
'is_permit': is_permit,
'is_ipv6': p.haslayer(IPv6),
'src_ip_addr': inet_pton(rule_family,
p[rule_l3_layer].src),
'src_ip_prefix_len': rule_prefix_len,
'dst_ip_addr': inet_pton(rule_family,
p[rule_l3_layer].dst),
'dst_ip_prefix_len': rule_prefix_len,
'srcport_or_icmptype_first': rule_l4_sport_first,
'srcport_or_icmptype_last': rule_l4_sport_last,
'dstport_or_icmpcode_first': rule_l4_dport,
'dstport_or_icmpcode_last': rule_l4_dport,
'proto': rule_l4_proto,
}
return new_rule
Packet.to_acl_rule = to_acl_rule
class IterateWithSleep():
def __init__(self, testcase, n_iters, description, sleep_sec):
self.curr = 0
self.testcase = testcase
self.n_iters = n_iters
self.sleep_sec = sleep_sec
self.description = description
def __iter__(self):
for x in range(0, self.n_iters):
yield x
self.testcase.sleep(self.sleep_sec)
class Conn(L4_Conn):
def apply_acls(self, reflect_side, acl_side):
pkts = []
pkts.append(self.pkt(0))
pkts.append(self.pkt(1))
pkt = pkts[reflect_side]
r = []
r.append(pkt.to_acl_rule(2, wildcard_sport=True))
r.append(self.wildcard_rule(0))
res = self.testcase.vapi.acl_add_replace(0xffffffff, r)
self.testcase.assert_equal(res.retval, 0, "error adding ACL")
reflect_acl_index = res.acl_index
r = []
r.append(self.wildcard_rule(0))
res = self.testcase.vapi.acl_add_replace(0xffffffff, r)
self.testcase.assert_equal(res.retval, 0, "error adding deny ACL")
deny_acl_index = res.acl_index
if reflect_side == acl_side:
self.testcase.vapi.acl_interface_set_acl_list(
self.ifs[acl_side].sw_if_index, 1,
[reflect_acl_index,
deny_acl_index])
self.testcase.vapi.acl_interface_set_acl_list(
self.ifs[1-acl_side].sw_if_index, 0, [])
else:
self.testcase.vapi.acl_interface_set_acl_list(
self.ifs[acl_side].sw_if_index, 1,
[deny_acl_index,
reflect_acl_index])
self.testcase.vapi.acl_interface_set_acl_list(
self.ifs[1-acl_side].sw_if_index, 0, [])
def wildcard_rule(self, is_permit):
any_addr = ["0.0.0.0", "::"]
rule_family = self.address_family
is_ip6 = 1 if rule_family == AF_INET6 else 0
new_rule = {
'is_permit': is_permit,
'is_ipv6': is_ip6,
'src_ip_addr': inet_pton(rule_family, any_addr[is_ip6]),
'src_ip_prefix_len': 0,
'dst_ip_addr': inet_pton(rule_family, any_addr[is_ip6]),
'dst_ip_prefix_len': 0,
'srcport_or_icmptype_first': 0,
'srcport_or_icmptype_last': 65535,
'dstport_or_icmpcode_first': 0,
'dstport_or_icmpcode_last': 65535,
'proto': 0,
}
return new_rule
@unittest.skipUnless(running_extended_tests(), "part of extended tests")
class ACLPluginConnTestCase(VppTestCase):
""" ACL plugin connection-oriented extended testcases """
@classmethod
def setUpClass(self):
super(ACLPluginConnTestCase, self).setUpClass()
# create pg0 and pg1
self.create_pg_interfaces(range(2))
cmd = "set acl-plugin session table event-trace 1"
self.logger.info(self.vapi.cli(cmd))
for i in self.pg_interfaces:
i.admin_up()
i.config_ip4()
i.config_ip6()
i.resolve_arp()
i.resolve_ndp()
def tearDown(self):
"""Run standard test teardown and log various show commands
"""
super(ACLPluginConnTestCase, self).tearDown()
if not self.vpp_dead:
self.logger.info(self.vapi.cli("show ip arp"))
self.logger.info(self.vapi.cli("show ip6 neighbors"))
self.logger.info(self.vapi.cli("show acl-plugin sessions"))
self.logger.info(self.vapi.cli("show acl-plugin acl"))
self.logger.info(self.vapi.cli("show acl-plugin interface"))
self.logger.info(self.vapi.cli("show acl-plugin tables"))
self.logger.info(self.vapi.cli("show event-logger all"))
def run_basic_conn_test(self, af, acl_side):
""" 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)
# the return packets should pass
conn1.send_through(1)
# send some packets on conn1, ensure it doesn't go away
for i in IterateWithSleep(self, 20, "Keep conn active", 0.3):
conn1.send_through(1)
# allow the conn to time out
for i in IterateWithSleep(self, 30, "Wait for timeout", 0.1):
pass
# now try to send a packet on the reflected side
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 long-idle conn")
def run_active_conn_test(self, af, 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)
conn1.apply_acls(0, acl_side)
conn1.send(0)
conn1.recv(1)
# create and check that the conn2/3 work
self.sleep(0.1)
conn2.send_pingpong(0)
self.sleep(0.1)
conn3.send_pingpong(0)
# send some packets on conn1, keep conn2/3 idle
for i in IterateWithSleep(self, 20, "Keep conn active", 0.2):
conn1.send_through(1)
try:
p2 = conn2.send_through(1).command()
except:
# If we asserted while waiting, it's good.
# the conn should have timed out.
p2 = None
# We should have not received the packet on a long-idle
# connection, because it should have timed out
# If it didn't - it is a problem
self.assert_equal(p2, None, "packet on long-idle conn")
def run_clear_conn_test(self, af, acl_side):
""" 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)
# the return packets should pass
conn1.send_through(1)
# send some packets on conn1, ensure it doesn't go away
for i in IterateWithSleep(self, 20, "Keep conn active", 0.3):
conn1.send_through(1)
# clear all connections
self.vapi.ppcli("clear acl-plugin sessions")
# now try to send a packet on the reflected side
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_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_0001_basic_conn_test(self):
""" 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 """
self.run_basic_conn_test(AF_INET, 1)
def test_0005_clear_conn_test(self):
""" 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 """
self.run_clear_conn_test(AF_INET, 0)
def test_0011_active_conn_test(self):
""" 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 """
self.run_active_conn_test(AF_INET, 1)
def test_1001_basic_conn_test(self):
""" 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 """
self.run_basic_conn_test(AF_INET6, 1)
def test_1005_clear_conn_test(self):
""" 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 """
self.run_clear_conn_test(AF_INET6, 0)
def test_1011_active_conn_test(self):
""" 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 """
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)
