#!/usr/bin/env python3 import socket import unittest from asfframework import VppTestRunner from scapy.packet import Raw from scapy.layers.l2 import Ether from scapy.layers.inet import IP, UDP, TCP from template_classifier import TestClassifier, VarMask, VarMatch from vpp_ip_route import VppIpRoute, VppRoutePath from vpp_ip import INVALID_INDEX from vpp_papi import VppEnum # Tests split to different test case classes because of issue reported in # ticket VPP-1336 class TestClassifierIP(TestClassifier): """Classifier IP Test Case""" @classmethod def setUpClass(cls): super(TestClassifierIP, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestClassifierIP, cls).tearDownClass() def test_iacl_src_ip(self): """Source IP iACL test Test scenario for basic IP ACL with source IP - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with source IP address. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with source IP pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes) self.pg0.add_stream(pkts) key = "ip_src" self.create_classify_table(key, self.build_ip_mask(src_ip="ffffffff")) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(src_ip=self.pg0.remote_ip4) ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_iacl_dst_ip(self): """Destination IP iACL test Test scenario for basic IP ACL with destination IP - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with destination IP address. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with destination IP pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes) self.pg0.add_stream(pkts) key = "ip_dst" self.create_classify_table(key, self.build_ip_mask(dst_ip="ffffffff")) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(dst_ip=self.pg1.remote_ip4) ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_iacl_src_dst_ip(self): """Source and destination IP iACL test Test scenario for basic IP ACL with source and destination IP - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with source and destination IP addresses. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with source and destination IP pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes) self.pg0.add_stream(pkts) key = "ip" self.create_classify_table( key, self.build_ip_mask(src_ip="ffffffff", dst_ip="ffffffff") ) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(src_ip=self.pg0.remote_ip4, dst_ip=self.pg1.remote_ip4), ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") class TestClassifierUDP(TestClassifier): """Classifier UDP proto Test Case""" @classmethod def setUpClass(cls): super(TestClassifierUDP, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestClassifierUDP, cls).tearDownClass() def test_iacl_proto_udp(self): """UDP protocol iACL test Test scenario for basic protocol ACL with UDP protocol - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with UDP IP protocol. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with UDP protocol pkts = self.create_stream(self.pg0, self.pg1, self.pg_if_packet_sizes) self.pg0.add_stream(pkts) key = "proto_udp" self.create_classify_table(key, self.build_ip_mask(proto="ff")) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(proto=socket.IPPROTO_UDP) ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_iacl_proto_udp_sport(self): """UDP source port iACL test Test scenario for basic protocol ACL with UDP and sport - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with UDP IP protocol and defined sport. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with UDP and sport sport = 38 pkts = self.create_stream( self.pg0, self.pg1, self.pg_if_packet_sizes, UDP(sport=sport, dport=5678) ) self.pg0.add_stream(pkts) key = "proto_udp_sport" self.create_classify_table(key, self.build_ip_mask(proto="ff", src_port="ffff")) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(proto=socket.IPPROTO_UDP, src_port=sport), ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_iacl_proto_udp_dport(self): """UDP destination port iACL test Test scenario for basic protocol ACL with UDP and dport - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with UDP IP protocol and defined dport. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with UDP and dport dport = 427 pkts = self.create_stream( self.pg0, self.pg1, self.pg_if_packet_sizes, UDP(sport=1234, dport=dport) ) self.pg0.add_stream(pkts) key = "proto_udp_dport" self.create_classify_table(key, self.build_ip_mask(proto="ff", dst_port="ffff")) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(proto=socket.IPPROTO_UDP, dst_port=dport), ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_iacl_proto_udp_sport_dport(self): """UDP source and destination ports iACL test Test scenario for basic protocol ACL with UDP and sport and dport - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with UDP IP protocol and defined sport and dport. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with UDP and sport and dport sport = 13720 dport = 9080 pkts = self.create_stream( self.pg0, self.pg1, self.pg_if_packet_sizes, UDP(sport=sport, dport=dport) ) self.pg0.add_stream(pkts) key = "proto_udp_ports" self.create_classify_table( key, self.build_ip_mask(proto="ff", src_port="ffff", dst_port="ffff") ) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match( proto=socket.IPPROTO_UDP, src_port=sport, dst_port=dport ), ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") class TestClassifierTCP(TestClassifier): """Classifier TCP proto Test Case""" @classmethod def setUpClass(cls): super(TestClassifierTCP, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestClassifierTCP, cls).tearDownClass() def test_iacl_proto_tcp(self): """TCP protocol iACL test Test scenario for basic protocol ACL with TCP protocol - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with TCP IP protocol. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with TCP protocol pkts = self.create_stream( self.pg0, self.pg1, self.pg_if_packet_sizes, TCP(sport=1234, dport=5678) ) self.pg0.add_stream(pkts) key = "proto_tcp" self.create_classify_table(key, self.build_ip_mask(proto="ff")) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(proto=socket.IPPROTO_TCP) ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts, TCP) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_iacl_proto_tcp_sport(self): """TCP source port iACL test Test scenario for basic protocol ACL with TCP and sport - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with TCP IP protocol and defined sport. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with TCP and sport sport = 38 pkts = self.create_stream( self.pg0, self.pg1, self.pg_if_packet_sizes, TCP(sport=sport, dport=5678) ) self.pg0.add_stream(pkts) key = "proto_tcp_sport" self.create_classify_table(key, self.build_ip_mask(proto="ff", src_port="ffff")) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(proto=socket.IPPROTO_TCP, src_port=sport), ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts, TCP) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_iacl_proto_tcp_dport(self): """TCP destination port iACL test Test scenario for basic protocol ACL with TCP and dport - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with TCP IP protocol and defined dport. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with TCP and dport dport = 427 pkts = self.create_stream( self.pg0, self.pg1, self.pg_if_packet_sizes, TCP(sport=1234, dport=dport) ) self.pg0.add_stream(pkts) key = "proto_tcp_sport" self.create_classify_table(key, self.build_ip_mask(proto="ff", dst_port="ffff")) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(proto=socket.IPPROTO_TCP, dst_port=dport), ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts, TCP) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_iacl_proto_tcp_sport_dport(self): """TCP source and destination ports iACL test Test scenario for basic protocol ACL with TCP and sport and dport - Create IPv4 stream for pg0 -> pg1 interface. - Create iACL with TCP IP protocol and defined sport and dport. - Send and verify received packets on pg1 interface. """ # Basic iACL testing with TCP and sport and dport sport = 13720 dport = 9080 pkts = self.create_stream( self.pg0, self.pg1, self.pg_if_packet_sizes, TCP(sport=sport, dport=dport) ) self.pg0.add_stream(pkts) key = "proto_tcp_ports" self.create_classify_table( key, self.build_ip_mask(proto="ff", src_port="ffff", dst_port="ffff") ) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match( proto=socket.IPPROTO_TCP, src_port=sport, dst_port=dport ), ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts, TCP) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") class TestClassifierIPOut(TestClassifier): """Classifier output IP Test Case""" @classmethod def setUpClass(cls): super(TestClassifierIPOut, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestClassifierIPOut, cls).tearDownClass() def test_acl_ip_out(self): """Output IP ACL test Test scenario for basic IP ACL with source IP - Create IPv4 stream for pg1 -> pg0 interface. - Create ACL with source IP address. - Send and verify received packets on pg0 interface. """ # Basic oACL testing with source IP pkts = self.create_stream(self.pg1, self.pg0, self.pg_if_packet_sizes) self.pg1.add_stream(pkts) key = "ip_out" self.create_classify_table( key, self.build_ip_mask(src_ip="ffffffff"), data_offset=0 ) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(src_ip=self.pg1.remote_ip4) ) self.output_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg0.get_capture(len(pkts)) self.verify_capture(self.pg0, pkts) self.pg1.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") class TestClassifierMAC(TestClassifier): """Classifier MAC Test Case""" @classmethod def setUpClass(cls): super(TestClassifierMAC, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestClassifierMAC, cls).tearDownClass() def test_acl_mac(self): """MAC ACL test Test scenario for basic MAC ACL with source MAC - Create IPv4 stream for pg0 -> pg2 interface. - Create ACL with source MAC address. - Send and verify received packets on pg2 interface. """ # Basic iACL testing with source MAC pkts = self.create_stream(self.pg0, self.pg2, self.pg_if_packet_sizes) self.pg0.add_stream(pkts) key = "mac" self.create_classify_table( key, self.build_mac_mask(src_mac="ffffffffffff"), data_offset=-14 ) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_mac_match(src_mac=self.pg0.remote_mac) ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg2.get_capture(len(pkts)) self.verify_capture(self.pg2, pkts) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg1.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") class TestClassifierComplex(TestClassifier): """Large & Nested Classifiers Test Cases""" @classmethod def setUpClass(cls): super(TestClassifierComplex, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestClassifierComplex, cls).tearDownClass() def test_iacl_large(self): """Large input ACL test Test scenario for Large ACL matching on ethernet+ip+udp headers - Create IPv4 stream for pg0 -> pg1 interface. - Create large acl matching on ethernet+ip+udp header fields - Send and verify received packets on pg1 interface. """ # 40b offset = 80bytes - (sizeof(UDP/IP/ETH) + 4b) # + 4b as build_ip_ma*() func, do not match against UDP Len & Chksum msk = VarMask(offset=40, spec="ffff") mth = VarMatch(offset=40, value=0x1234, length=2) payload_msk = self.build_payload_mask([msk]) payload_match = self.build_payload_match([mth]) sport = 13720 dport = 9080 # 36b offset = 80bytes - (sizeof(UDP/IP/ETH)) packet_ex = bytes.fromhex(("0" * 36) + "1234") pkts = self.create_stream( self.pg0, self.pg1, self.pg_if_packet_sizes, UDP(sport=sport, dport=dport), packet_ex, ) self.pg0.add_stream(pkts) key = "large_in" self.create_classify_table( key, self.build_mac_mask( src_mac="ffffffffffff", dst_mac="ffffffffffff", ether_type="ffff" ) + self.build_ip_mask( proto="ff", src_ip="ffffffff", dst_ip="ffffffff", src_port="ffff", dst_port="ffff", ) + payload_msk, data_offset=-14, ) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_mac_match( src_mac=self.pg0.remote_mac, dst_mac=self.pg0.local_mac, # ipv4 next header ether_type="0800", ) + self.build_ip_match( proto=socket.IPPROTO_UDP, src_ip=self.pg0.remote_ip4, dst_ip=self.pg1.remote_ip4, src_port=sport, dst_port=dport, ) + payload_match, ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_oacl_large(self): """Large output ACL test Test scenario for Large ACL matching on ethernet+ip+udp headers - Create IPv4 stream for pg1 -> pg0 interface. - Create large acl matching on ethernet+ip+udp header fields - Send and verify received packets on pg0 interface. """ # 40b offset = 80bytes - (sizeof(UDP/IP/ETH) + 4b) # + 4b as build_ip_ma*() func, do not match against UDP Len & Chksum msk = VarMask(offset=40, spec="ffff") mth = VarMatch(offset=40, value=0x1234, length=2) payload_msk = self.build_payload_mask([msk]) payload_match = self.build_payload_match([mth]) sport = 13720 dport = 9080 # 36b offset = 80bytes - (sizeof(UDP/IP/ETH)) packet_ex = bytes.fromhex(("0" * 36) + "1234") pkts = self.create_stream( self.pg1, self.pg0, self.pg_if_packet_sizes, UDP(sport=sport, dport=dport), packet_ex, ) self.pg1.add_stream(pkts) key = "large_out" self.create_classify_table( key, self.build_mac_mask( src_mac="ffffffffffff", dst_mac="ffffffffffff", ether_type="ffff" ) + self.build_ip_mask( proto="ff", src_ip="ffffffff", dst_ip="ffffffff", src_port="ffff", dst_port="ffff", ) + payload_msk, data_offset=-14, ) self.create_classify_session( self.acl_tbl_idx.get(key), self.build_mac_match( src_mac=self.pg0.local_mac, dst_mac=self.pg0.remote_mac, # ipv4 next header ether_type="0800", ) + self.build_ip_match( proto=socket.IPPROTO_UDP, src_ip=self.pg1.remote_ip4, dst_ip=self.pg0.remote_ip4, src_port=sport, dst_port=dport, ) + payload_match, ) self.output_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg0.get_capture(len(pkts)) self.verify_capture(self.pg0, pkts) self.pg1.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_iacl_nested(self): """Nested input ACL test Test scenario for Large ACL matching on ethernet+ip+udp headers - Create IPv4 stream for pg0 -> pg1 interface. - Create 1st classifier table, without any entries - Create nested acl matching on ethernet+ip+udp header fields - Send and verify received packets on pg1 interface. """ sport = 13720 dport = 9080 pkts = self.create_stream( self.pg0, self.pg1, self.pg_if_packet_sizes, UDP(sport=sport, dport=dport) ) self.pg0.add_stream(pkts) subtable_key = "subtable_in" self.create_classify_table( subtable_key, self.build_mac_mask( src_mac="ffffffffffff", dst_mac="ffffffffffff", ether_type="ffff" ) + self.build_ip_mask( proto="ff", src_ip="ffffffff", dst_ip="ffffffff", src_port="ffff", dst_port="ffff", ), data_offset=-14, ) key = "nested_in" self.create_classify_table( key, self.build_mac_mask( src_mac="ffffffffffff", dst_mac="ffffffffffff", ether_type="ffff" ) + self.build_ip_mask( proto="ff", src_ip="ffffffff", dst_ip="ffffffff", src_port="ffff", dst_port="ffff", ), next_table_index=self.acl_tbl_idx.get(subtable_key), ) self.create_classify_session( self.acl_tbl_idx.get(subtable_key), self.build_mac_match( src_mac=self.pg0.remote_mac, dst_mac=self.pg0.local_mac, # ipv4 next header ether_type="0800", ) + self.build_ip_match( proto=socket.IPPROTO_UDP, src_ip=self.pg0.remote_ip4, dst_ip=self.pg1.remote_ip4, src_port=sport, dst_port=dport, ), ) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg1.get_capture(len(pkts)) self.verify_capture(self.pg1, pkts) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") def test_oacl_nested(self): """Nested output ACL test Test scenario for Large ACL matching on ethernet+ip+udp headers - Create IPv4 stream for pg1 -> pg0 interface. - Create 1st classifier table, without any entries - Create nested acl matching on ethernet+ip+udp header fields - Send and verify received packets on pg0 interface. """ sport = 13720 dport = 9080 pkts = self.create_stream( self.pg1, self.pg0, self.pg_if_packet_sizes, UDP(sport=sport, dport=dport) ) self.pg1.add_stream(pkts) subtable_key = "subtable_out" self.create_classify_table( subtable_key, self.build_mac_mask( src_mac="ffffffffffff", dst_mac="ffffffffffff", ether_type="ffff" ) + self.build_ip_mask( proto="ff", src_ip="ffffffff", dst_ip="ffffffff", src_port="ffff", dst_port="ffff", ), data_offset=-14, ) key = "nested_out" self.create_classify_table( key, self.build_mac_mask( src_mac="ffffffffffff", dst_mac="ffffffffffff", ether_type="ffff" ) + self.build_ip_mask( proto="ff", src_ip="ffffffff", dst_ip="ffffffff", src_port="ffff", dst_port="ffff", ), next_table_index=self.acl_tbl_idx.get(subtable_key), data_offset=-14, ) self.create_classify_session( self.acl_tbl_idx.get(subtable_key), self.build_mac_match( src_mac=self.pg0.local_mac, dst_mac=self.pg0.remote_mac, # ipv4 next header ether_type="0800", ) + self.build_ip_match( proto=socket.IPPROTO_UDP, src_ip=self.pg1.remote_ip4, dst_ip=self.pg0.remote_ip4, src_port=sport, dst_port=dport, ), ) self.output_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg0.get_capture(len(pkts)) self.verify_capture(self.pg0, pkts) self.pg1.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") self.pg3.assert_nothing_captured(remark="packets forwarded") class TestClassifierPBR(TestClassifier): """Classifier PBR Test Case""" @classmethod def setUpClass(cls): super(TestClassifierPBR, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestClassifierPBR, cls).tearDownClass() def test_acl_pbr(self): """IP PBR test Test scenario for PBR with source IP - Create IPv4 stream for pg0 -> pg3 interface. - Configure PBR fib entry for packet forwarding. - Send and verify received packets on pg3 interface. """ # PBR testing with source IP pkts = self.create_stream(self.pg0, self.pg3, self.pg_if_packet_sizes) self.pg0.add_stream(pkts) key = "pbr" self.create_classify_table(key, self.build_ip_mask(src_ip="ffffffff")) pbr_option = 1 # this will create the VRF/table in which we will insert the route self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(src_ip=self.pg0.remote_ip4), pbr_option, self.pbr_vrfid, ) self.assertTrue(self.verify_vrf(self.pbr_vrfid)) r = VppIpRoute( self, self.pg3.local_ip4, 24, [VppRoutePath(self.pg3.remote_ip4, INVALID_INDEX)], table_id=self.pbr_vrfid, ) r.add_vpp_config() self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.pg_enable_capture(self.pg_interfaces) self.pg_start() pkts = self.pg3.get_capture(len(pkts)) self.verify_capture(self.pg3, pkts) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key), 0) self.pg0.assert_nothing_captured(remark="packets forwarded") self.pg1.assert_nothing_captured(remark="packets forwarded") self.pg2.assert_nothing_captured(remark="packets forwarded") # remove the classify session and the route r.remove_vpp_config() self.create_classify_session( self.acl_tbl_idx.get(key), self.build_ip_match(src_ip=self.pg0.remote_ip4), pbr_option, self.pbr_vrfid, is_add=0, ) # and the table should be gone. self.assertFalse(self.verify_vrf(self.pbr_vrfid)) class TestClassifierPunt(TestClassifier): """Classifier punt Test Case""" @classmethod def setUpClass(cls): super(TestClassifierPunt, cls).setUpClass() @classmethod def tearDownClass(cls): super(TestClassifierPunt, cls).tearDownClass() def test_punt_udp(self): """IPv4/UDP protocol punt ACL test Test scenario for basic punt ACL with UDP protocol - Create IPv4 stream for pg0 -> pg1 interface. - Create punt ACL with UDP IP protocol. - Send and verify received packets on pg1 interface. """ sport = 6754 dport = 17923 key = "ip4_udp_punt" self.create_classify_table( key, self.build_ip_mask(src_ip="ffffffff", proto="ff", src_port="ffff") ) table_index = self.acl_tbl_idx.get(key) self.vapi.punt_acl_add_del(ip4_table_index=table_index) self.acl_active_table = key # punt udp packets to dport received on pg0 through pg1 self.vapi.set_punt( is_add=1, punt={ "type": VppEnum.vl_api_punt_type_t.PUNT_API_TYPE_L4, "punt": { "l4": { "af": VppEnum.vl_api_address_family_t.ADDRESS_IP4, "protocol": VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP, "port": dport, } }, }, ) self.vapi.ip_punt_redirect( punt={ "rx_sw_if_index": self.pg0.sw_if_index, "tx_sw_if_index": self.pg1.sw_if_index, "nh": self.pg1.remote_ip4, } ) pkts = [ ( Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) / UDP(sport=sport, dport=dport) / Raw("\x17" * 100) ) ] * 2 # allow a session but not matching the stream: expect to drop self.create_classify_session( table_index, self.build_ip_match( src_ip=self.pg0.remote_ip4, proto=socket.IPPROTO_UDP, src_port=sport + 10, ), ) self.send_and_assert_no_replies(self.pg0, pkts) # allow a session matching the stream: expect to pass self.create_classify_session( table_index, self.build_ip_match( src_ip=self.pg0.remote_ip4, proto=socket.IPPROTO_UDP, src_port=sport ), ) self.send_and_expect_only(self.pg0, pkts, self.pg1) # test dump api: ip4 is set, ip6 is not r = self.vapi.punt_acl_get() self.assertEqual(r.ip4_table_index, table_index) self.assertEqual(r.ip6_table_index, 0xFFFFFFFF) # cleanup self.acl_active_table = "" self.vapi.punt_acl_add_del(ip4_table_index=table_index, is_add=0) # test dump api: nothing set r = self.vapi.punt_acl_get() self.assertEqual(r.ip4_table_index, 0xFFFFFFFF) self.assertEqual(r.ip6_table_index, 0xFFFFFFFF) if __name__ == "__main__": unittest.main(testRunner=VppTestRunner)