Use consistent API types.
Type: fix
Signed-off-by: Jakub Grajciar <jgrajcia@cisco.com>
Change-Id: If90d753f129312400c4c3669bb86289d0c3e0d99
Signed-off-by: Jakub Grajciar <jgrajcia@cisco.com>
l2_list::l2_list(const key_t& key)
: m_hdl(handle_t::INVALID)
, m_key(key)
-{
-}
+{}
l2_list::l2_list(const handle_t& hdl, const key_t& key)
: m_hdl(hdl)
, m_key(key)
-{
-}
+{}
l2_list::l2_list(const key_t& key, const rules_t& rules)
: m_hdl(handle_t::INVALID)
, m_key(key)
, m_rules(rules)
-{
-}
+{}
l2_list::l2_list(const l2_list& o)
: m_hdl(o.m_hdl)
, m_key(o.m_key)
, m_rules(o.m_rules)
-{
-}
+{}
l2_list::~l2_list()
{
l2_list acl(hdl, std::string(reinterpret_cast<const char*>(payload.tag)));
for (unsigned int ii = 0; ii < payload.count; ii++) {
- const route::prefix_t pfx(payload.r[ii].is_ipv6,
- payload.r[ii].src_ip_addr,
- payload.r[ii].src_ip_prefix_len);
- l2_rule rule(ii, action_t::from_int(payload.r[ii].is_permit), pfx,
- { payload.r[ii].src_mac }, { payload.r[ii].src_mac_mask });
+ const route::prefix_t pfx(payload.r[ii].src_prefix.address.af,
+ (uint8_t*)&payload.r[ii].src_prefix.address.un,
+ payload.r[ii].src_prefix.len);
+ l2_rule rule(ii,
+ action_t::from_int(payload.r[ii].is_permit),
+ pfx,
+ { payload.r[ii].src_mac },
+ { payload.r[ii].src_mac_mask });
acl.insert(rule);
}
l3_list::l3_list(const key_t& key)
: m_hdl(handle_t::INVALID)
, m_key(key)
-{
-}
+{}
l3_list::l3_list(const handle_t& hdl, const key_t& key)
: m_hdl(hdl)
, m_key(key)
-{
-}
+{}
l3_list::l3_list(const key_t& key, const rules_t& rules)
: m_hdl(handle_t::INVALID)
, m_key(key)
, m_rules(rules)
-{
-}
+{}
l3_list::l3_list(const l3_list& o)
: m_hdl(o.m_hdl)
, m_key(o.m_key)
, m_rules(o.m_rules)
-{
-}
+{}
l3_list::~l3_list()
{
l3_list acl(hdl, std::string(reinterpret_cast<const char*>(payload.tag)));
for (unsigned int ii = 0; ii < payload.count; ii++) {
- const route::prefix_t src(payload.r[ii].is_ipv6,
- payload.r[ii].src_ip_addr,
- payload.r[ii].src_ip_prefix_len);
- const route::prefix_t dst(payload.r[ii].is_ipv6,
- payload.r[ii].dst_ip_addr,
- payload.r[ii].dst_ip_prefix_len);
+ const route::prefix_t src(payload.r[ii].src_prefix.address.af,
+ (uint8_t*)&payload.r[ii].src_prefix.address.un,
+ payload.r[ii].src_prefix.len);
+ const route::prefix_t dst(payload.r[ii].dst_prefix.address.af,
+ (uint8_t*)&payload.r[ii].dst_prefix.address.un,
+ payload.r[ii].dst_prefix.len);
l3_rule rule(ii, action_t::from_int(payload.r[ii].is_permit), src, dst);
rule.set_proto(payload.r[ii].proto);
static void
to_vpp(const l2_rule& rule, vapi_type_macip_acl_rule& payload)
{
- payload.is_permit = rule.action().value();
- rule.src_ip().to_vpp(&payload.is_ipv6, payload.src_ip_addr,
- &payload.src_ip_prefix_len);
+ payload.is_permit = (vapi_enum_acl_action)rule.action().value();
+ rule.src_ip().to_vpp((uint8_t*)&payload.src_prefix.address.af,
+ (uint8_t*)&payload.src_prefix.address.un,
+ &payload.src_prefix.len);
rule.mac().to_bytes(payload.src_mac, 6);
rule.mac_mask().to_bytes(payload.src_mac_mask, 6);
}
static void
to_vpp(const l3_rule& rule, vapi_type_acl_rule& payload)
{
- payload.is_permit = rule.action().value();
- rule.src().to_vpp(&payload.is_ipv6, payload.src_ip_addr,
- &payload.src_ip_prefix_len);
- rule.dst().to_vpp(&payload.is_ipv6, payload.dst_ip_addr,
- &payload.dst_ip_prefix_len);
-
- payload.proto = rule.proto();
+ payload.is_permit = (vapi_enum_acl_action)rule.action().value();
+ rule.src().to_vpp((uint8_t*)&payload.src_prefix.address.af,
+ (uint8_t*)&payload.src_prefix.address.un,
+ &payload.src_prefix.len);
+ rule.dst().to_vpp((uint8_t*)&payload.dst_prefix.address.af,
+ (uint8_t*)&payload.dst_prefix.address.un,
+ &payload.dst_prefix.len);
+
+ payload.proto = (vapi_enum_ip_proto)rule.proto();
payload.srcport_or_icmptype_first = rule.srcport_or_icmptype_first();
payload.srcport_or_icmptype_last = rule.srcport_or_icmptype_last();
payload.dstport_or_icmpcode_first = rule.dstport_or_icmpcode_first();
payload.tcp_flags_value = rule.tcp_flags_value();
}
-template <>
+template<>
rc_t
l3_update_cmd::issue(connection& con)
{
payload.acl_index = m_hw_item.data().value();
payload.count = m_rules.size();
memset(payload.tag, 0, sizeof(payload.tag));
- memcpy(payload.tag, m_key.c_str(),
- std::min(m_key.length(), sizeof(payload.tag)));
+ memcpy(
+ payload.tag, m_key.c_str(), std::min(m_key.length(), sizeof(payload.tag)));
auto it = m_rules.cbegin();
return rc_t::OK;
}
-template <>
+template<>
rc_t
l3_delete_cmd::issue(connection& con)
{
return rc_t::OK;
}
-template <>
+template<>
rc_t
l3_dump_cmd::issue(connection& con)
{
return rc_t::OK;
}
-template <>
+template<>
rc_t
l2_update_cmd::issue(connection& con)
{
// payload.acl_index = m_hw_item.data().value();
payload.count = m_rules.size();
memset(payload.tag, 0, sizeof(payload.tag));
- memcpy(payload.tag, m_key.c_str(),
- std::min(m_key.length(), sizeof(payload.tag)));
+ memcpy(
+ payload.tag, m_key.c_str(), std::min(m_key.length(), sizeof(payload.tag)));
auto it = m_rules.cbegin();
return rc_t::OK;
}
-template <>
+template<>
rc_t
l2_delete_cmd::issue(connection& con)
{
return rc_t::OK;
}
-template <>
+template<>
rc_t
l2_dump_cmd::issue(connection& con)
{
from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsLabel, \
VppIpTable, FibPathProto
+from vpp_acl import AclRule, VppAcl
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
+from ipaddress import IPv4Network, IPv6Network
from vpp_object import VppObject
#
# Rule 1
#
- rule_1 = ({'is_permit': 1,
- 'is_ipv6': 0,
- 'proto': 17,
- 'srcport_or_icmptype_first': 1234,
- 'srcport_or_icmptype_last': 1234,
- 'src_ip_prefix_len': 32,
- 'src_ip_addr': inet_pton(AF_INET, "1.1.1.1"),
- 'dstport_or_icmpcode_first': 1234,
- 'dstport_or_icmpcode_last': 1234,
- 'dst_ip_prefix_len': 32,
- 'dst_ip_addr': inet_pton(AF_INET, "1.1.1.2")})
- acl_1 = self.vapi.acl_add_replace(acl_index=4294967295, r=[rule_1])
+ rule_1 = AclRule(is_permit=1, proto=17, ports=1234,
+ src_prefix=IPv4Network("1.1.1.1/32"),
+ dst_prefix=IPv4Network("1.1.1.2/32"))
+ acl_1 = VppAcl(self, rules=[rule_1])
+ acl_1.add_vpp_config()
#
# ABF policy for ACL 1 - path via interface 1
#
# Rule 1
#
- rule_1 = ({'is_permit': 1,
- 'is_ipv6': 1,
- 'proto': 17,
- 'srcport_or_icmptype_first': 1234,
- 'srcport_or_icmptype_last': 1234,
- 'src_ip_prefix_len': 128,
- 'src_ip_addr': inet_pton(AF_INET6, "2001::2"),
- 'dstport_or_icmpcode_first': 1234,
- 'dstport_or_icmpcode_last': 1234,
- 'dst_ip_prefix_len': 128,
- 'dst_ip_addr': inet_pton(AF_INET6, "2001::1")})
- acl_1 = self.vapi.acl_add_replace(acl_index=4294967295,
- r=[rule_1])
+ rule_1 = AclRule(is_permit=1, proto=17, ports=1234,
+ src_prefix=IPv6Network("2001::2/128"),
+ dst_prefix=IPv6Network("2001::1/128"))
+ acl_1 = VppAcl(self, rules=[rule_1])
+ acl_1.add_vpp_config()
#
# ABF policy for ACL 1 - path via interface 1
used to control the ACL plugin
*/
-option version = "1.0.1";
+option version = "2.0.0";
import "plugins/acl/acl_types.api";
+import "vnet/interface_types.api";
/** \brief Get the plugin version
@param client_index - opaque cookie to identify the sender
u32 client_index;
u32 context;
u32 acl_index; /* ~0 to add, existing ACL# to replace */
- u8 tag[64]; /* What gets in here gets out in the corresponding tag field when dumping the ACLs. */
+ string tag[64]; /* What gets in here gets out in the corresponding tag field when dumping the ACLs. */
u32 count;
vl_api_acl_rule_t r[count];
option vat_help = "<acl-idx> [<ipv4|ipv6>] <permit|permit+reflect|deny|action N> [src IP/plen] [dst IP/plen] [sport X-Y] [dport X-Y] [proto P] [tcpflags FL MASK], ... , ...";
{
u32 client_index;
u32 context;
- u8 is_add;
+ bool is_add [default=true];
/*
* is_input = 0 => ACL applied on interface egress
* is_input = 1 => ACL applied on interface ingress
*/
- u8 is_input;
- u32 sw_if_index;
+ bool is_input;
+ vl_api_interface_index_t sw_if_index;
u32 acl_index;
option vat_help = "<intfc> | sw_if_index <if-idx> [add|del] [input|output] acl <acl-idx>";
};
{
u32 client_index;
u32 context;
- u32 sw_if_index;
+ vl_api_interface_index_t sw_if_index;
u8 count;
u8 n_input; /* First n_input ACLs are set as a list of input ACLs, the rest are applied as output */
u32 acls[count];
{
u32 context;
u32 acl_index;
- u8 tag[64]; /* Same blob that was supplied to us when creating the ACL, one hopes. */
+ string tag[64]; /* Same blob that was supplied to us when creating the ACL, one hopes. */
u32 count;
vl_api_acl_rule_t r[count];
};
{
u32 client_index;
u32 context;
- u32 sw_if_index; /* ~0 for all interfaces */
+ vl_api_interface_index_t sw_if_index; /* ~0 for all interfaces */
option vat_help = "[<intfc> | sw_if_index <if-idx>]";
};
define acl_interface_list_details
{
u32 context;
- u32 sw_if_index;
+ vl_api_interface_index_t sw_if_index;
u8 count;
u8 n_input;
u32 acls[count];
{
u32 client_index;
u32 context;
- u8 tag[64];
+ string tag[64];
u32 count;
vl_api_macip_acl_rule_t r[count];
option vat_help = "...";
u32 client_index;
u32 context;
u32 acl_index; /* ~0 to add, existing MACIP ACL# to replace */
- u8 tag[64];
+ string tag[64];
u32 count;
vl_api_macip_acl_rule_t r[count];
option vat_help = "<acl-idx> [<ipv4|ipv6>] <permit|deny|action N> [count <count>] [src] ip <ipaddress/[plen]> mac <mac> mask <mac_mask>, ... , ...";
{
u32 client_index;
u32 context;
- u8 is_add;
+ bool is_add [default=true];
/* MACIP ACLs are always input */
- u32 sw_if_index;
+ vl_api_interface_index_t sw_if_index;
u32 acl_index;
option vat_help = "<intfc> | sw_if_index <if-idx> [add|del] acl <acl-idx>";
};
{
u32 context;
u32 acl_index;
- u8 tag[64];
+ string tag[64];
u32 count;
vl_api_macip_acl_rule_t r[count];
};
{
u32 client_index;
u32 context;
- u32 sw_if_index; /* ~0 for all interfaces */
+ vl_api_interface_index_t sw_if_index; /* ~0 for all interfaces */
};
/** \brief Details about a single MACIP ACL contents
define macip_acl_interface_list_details
{
u32 context;
- u32 sw_if_index;
+ vl_api_interface_index_t sw_if_index;
u8 count;
u32 acls[count];
};
{
u32 client_index;
u32 context;
- u32 sw_if_index;
+ vl_api_interface_index_t sw_if_index;
u8 count; /* Total number of ethertypes in the whitelist */
u8 n_input; /* first n_input ethertypes are input, the rest - output */
u16 whitelist[count];
{
u32 client_index;
u32 context;
- u32 sw_if_index; /* ~0 for all interfaces */
+ vl_api_interface_index_t sw_if_index; /* ~0 for all interfaces */
option vat_help = "[<intfc> | sw_if_index <if-idx>]";
};
define acl_interface_etype_whitelist_details
{
u32 context;
- u32 sw_if_index;
+ vl_api_interface_index_t sw_if_index;
u8 count;
u8 n_input; /* first n_input ethertypes are input, the rest - output */
u16 whitelist[count];
#include <vnet/classify/in_out_acl.h>
#include <vpp/app/version.h>
+#include <vnet/ethernet/ethernet_types_api.h>
+
#include <vlibapi/api.h>
#include <vlibmemory/api.h>
}
static int
-acl_api_ip4_invalid_prefix (void *ip4_pref_raw, u8 ip4_prefix_len)
+acl_api_ip4_invalid_prefix (const vl_api_prefix_t * prefix)
{
ip4_address_t ip4_addr;
ip4_address_t ip4_mask;
ip4_address_t ip4_masked_addr;
- memcpy (&ip4_addr, ip4_pref_raw, sizeof (ip4_addr));
- ip4_preflen_to_mask (ip4_prefix_len, &ip4_mask);
+ if (prefix->len > 32)
+ return 1;
+
+ ip4_address_decode (prefix->address.un.ip4, &ip4_addr);
+ ip4_preflen_to_mask (prefix->len, &ip4_mask);
ip4_masked_addr.as_u32 = ip4_addr.as_u32 & ip4_mask.as_u32;
int ret = (ip4_masked_addr.as_u32 != ip4_addr.as_u32);
if (ret)
{
clib_warning
("inconsistent addr %U for prefix len %d; (%U when masked)",
- format_ip4_address, ip4_pref_raw, ip4_prefix_len, format_ip4_address,
- &ip4_masked_addr);
+ format_ip4_address, prefix->address.un.ip4, prefix->len,
+ format_ip4_address, &ip4_masked_addr);
}
return ret;
}
static int
-acl_api_ip6_invalid_prefix (void *ip6_pref_raw, u8 ip6_prefix_len)
+acl_api_ip6_invalid_prefix (const vl_api_prefix_t * prefix)
{
ip6_address_t ip6_addr;
ip6_address_t ip6_mask;
ip6_address_t ip6_masked_addr;
- memcpy (&ip6_addr, ip6_pref_raw, sizeof (ip6_addr));
- ip6_preflen_to_mask (ip6_prefix_len, &ip6_mask);
+ if (prefix->len > 128)
+ return 1;
+
+ ip6_address_decode (prefix->address.un.ip6, &ip6_addr);
+ ip6_preflen_to_mask (prefix->len, &ip6_mask);
ip6_masked_addr.as_u64[0] = ip6_addr.as_u64[0] & ip6_mask.as_u64[0];
ip6_masked_addr.as_u64[1] = ip6_addr.as_u64[1] & ip6_mask.as_u64[1];
int ret = ((ip6_masked_addr.as_u64[0] != ip6_addr.as_u64[0])
{
clib_warning
("inconsistent addr %U for prefix len %d; (%U when masked)",
- format_ip6_address, ip6_pref_raw, ip6_prefix_len, format_ip6_address,
- &ip6_masked_addr);
+ format_ip6_address, prefix->address.un.ip6, prefix->len,
+ format_ip6_address, &ip6_masked_addr);
}
return ret;
}
+static int
+acl_api_invalid_prefix (const vl_api_prefix_t * prefix)
+{
+ if (prefix->address.af == ADDRESS_IP6)
+ return acl_api_ip6_invalid_prefix (prefix);
+ return acl_api_ip4_invalid_prefix (prefix);
+}
+
static int
acl_add_list (u32 count, vl_api_acl_rule_t rules[],
u32 * acl_list_index, u8 * tag)
/* check if what they request is consistent */
for (i = 0; i < count; i++)
{
- if (rules[i].is_ipv6)
- {
- if (rules[i].src_ip_prefix_len > 128)
- return VNET_API_ERROR_INVALID_VALUE;
- if (rules[i].dst_ip_prefix_len > 128)
- return VNET_API_ERROR_INVALID_VALUE;
- if (acl_api_ip6_invalid_prefix
- (&rules[i].src_ip_addr, rules[i].src_ip_prefix_len))
- return VNET_API_ERROR_INVALID_SRC_ADDRESS;
- if (acl_api_ip6_invalid_prefix
- (&rules[i].dst_ip_addr, rules[i].dst_ip_prefix_len))
- return VNET_API_ERROR_INVALID_DST_ADDRESS;
- }
- else
- {
- if (rules[i].src_ip_prefix_len > 32)
- return VNET_API_ERROR_INVALID_VALUE;
- if (rules[i].dst_ip_prefix_len > 32)
- return VNET_API_ERROR_INVALID_VALUE;
- if (acl_api_ip4_invalid_prefix
- (&rules[i].src_ip_addr, rules[i].src_ip_prefix_len))
- return VNET_API_ERROR_INVALID_SRC_ADDRESS;
- if (acl_api_ip4_invalid_prefix
- (&rules[i].dst_ip_addr, rules[i].dst_ip_prefix_len))
- return VNET_API_ERROR_INVALID_DST_ADDRESS;
- }
+ if (acl_api_invalid_prefix (&rules[i].src_prefix))
+ return VNET_API_ERROR_INVALID_SRC_ADDRESS;
+ if (acl_api_invalid_prefix (&rules[i].dst_prefix))
+ return VNET_API_ERROR_INVALID_DST_ADDRESS;
if (ntohs (rules[i].srcport_or_icmptype_first) >
ntohs (rules[i].srcport_or_icmptype_last))
return VNET_API_ERROR_INVALID_VALUE_2;
r = vec_elt_at_index (acl_new_rules, i);
clib_memset (r, 0, sizeof (*r));
r->is_permit = rules[i].is_permit;
- r->is_ipv6 = rules[i].is_ipv6;
- if (r->is_ipv6)
- {
- memcpy (&r->src, rules[i].src_ip_addr, sizeof (r->src));
- memcpy (&r->dst, rules[i].dst_ip_addr, sizeof (r->dst));
- }
- else
- {
- memcpy (&r->src.ip4, rules[i].src_ip_addr, sizeof (r->src.ip4));
- memcpy (&r->dst.ip4, rules[i].dst_ip_addr, sizeof (r->dst.ip4));
- }
- r->src_prefixlen = rules[i].src_ip_prefix_len;
- r->dst_prefixlen = rules[i].dst_ip_prefix_len;
+ r->is_ipv6 = rules[i].src_prefix.address.af;
+ ip_address_decode (&rules[i].src_prefix.address, &r->src);
+ ip_address_decode (&rules[i].dst_prefix.address, &r->dst);
+ r->src_prefixlen = rules[i].src_prefix.len;
+ r->dst_prefixlen = rules[i].dst_prefix.len;
r->proto = rules[i].proto;
r->src_port_or_type_first = ntohs (rules[i].srcport_or_icmptype_first);
r->src_port_or_type_last = ntohs (rules[i].srcport_or_icmptype_last);
{
r = &acl_new_rules[i];
r->is_permit = rules[i].is_permit;
- r->is_ipv6 = rules[i].is_ipv6;
- memcpy (&r->src_mac, rules[i].src_mac, 6);
- memcpy (&r->src_mac_mask, rules[i].src_mac_mask, 6);
- if (rules[i].is_ipv6)
- memcpy (&r->src_ip_addr.ip6, rules[i].src_ip_addr, 16);
- else
- memcpy (&r->src_ip_addr.ip4, rules[i].src_ip_addr, 4);
- r->src_prefixlen = rules[i].src_ip_prefix_len;
+ r->is_ipv6 = rules[i].src_prefix.address.af;
+ mac_address_decode (rules[i].src_mac, (mac_address_t *) & r->src_mac);
+ mac_address_decode (rules[i].src_mac_mask,
+ (mac_address_t *) & r->src_mac_mask);
+ ip_address_decode (&rules[i].src_prefix.address, &r->src_ip_addr);
+ r->src_prefixlen = rules[i].src_prefix.len;
}
if (~0 == *acl_list_index)
copy_acl_rule_to_api_rule (vl_api_acl_rule_t * api_rule, acl_rule_t * r)
{
api_rule->is_permit = r->is_permit;
- api_rule->is_ipv6 = r->is_ipv6;
- if (r->is_ipv6)
- {
- memcpy (api_rule->src_ip_addr, &r->src, sizeof (r->src));
- memcpy (api_rule->dst_ip_addr, &r->dst, sizeof (r->dst));
- }
- else
- {
- memcpy (api_rule->src_ip_addr, &r->src.ip4, sizeof (r->src.ip4));
- memcpy (api_rule->dst_ip_addr, &r->dst.ip4, sizeof (r->dst.ip4));
- }
- api_rule->src_ip_prefix_len = r->src_prefixlen;
- api_rule->dst_ip_prefix_len = r->dst_prefixlen;
+ ip_address_encode (&r->src, r->is_ipv6 ? IP46_TYPE_IP6 : IP46_TYPE_IP4,
+ &api_rule->src_prefix.address);
+ ip_address_encode (&r->dst, r->is_ipv6 ? IP46_TYPE_IP6 : IP46_TYPE_IP4,
+ &api_rule->dst_prefix.address);
+ api_rule->src_prefix.len = r->src_prefixlen;
+ api_rule->dst_prefix.len = r->dst_prefixlen;
api_rule->proto = r->proto;
api_rule->srcport_or_icmptype_first = htons (r->src_port_or_type_first);
api_rule->srcport_or_icmptype_last = htons (r->src_port_or_type_last);
{
r = &acl->rules[i];
rules[i].is_permit = r->is_permit;
- rules[i].is_ipv6 = r->is_ipv6;
- memcpy (rules[i].src_mac, &r->src_mac, sizeof (r->src_mac));
- memcpy (rules[i].src_mac_mask, &r->src_mac_mask,
- sizeof (r->src_mac_mask));
- if (r->is_ipv6)
- memcpy (rules[i].src_ip_addr, &r->src_ip_addr.ip6,
- sizeof (r->src_ip_addr.ip6));
- else
- memcpy (rules[i].src_ip_addr, &r->src_ip_addr.ip4,
- sizeof (r->src_ip_addr.ip4));
- rules[i].src_ip_prefix_len = r->src_prefixlen;
+ mac_address_encode ((mac_address_t *) & r->src_mac,
+ rules[i].src_mac);
+ mac_address_encode ((mac_address_t *) & r->src_mac_mask,
+ rules[i].src_mac_mask);
+ ip_address_encode (&r->src_ip_addr,
+ r->is_ipv6 ? IP46_TYPE_IP6 : IP46_TYPE_IP4,
+ &rules[i].src_prefix.address);
+ rules[i].src_prefix.len = r->src_prefixlen;
}
}
else
#include <vnet/ip/ip.h>
#include <arpa/inet.h>
+#include <vnet/ip/ip_format_fns.h>
+#include <vnet/ethernet/ethernet_format_fns.h>
+
#define __plugin_msg_base acl_test_main.msg_id_base
#include <vlibapi/vat_helper_macros.h>
static inline u8 *
vl_api_acl_rule_t_pretty_format (u8 *out, vl_api_acl_rule_t * a)
{
- int af = a->is_ipv6 ? AF_INET6 : AF_INET;
+ int af = a->src_prefix.address.af ? AF_INET6 : AF_INET;
u8 src[INET6_ADDRSTRLEN];
u8 dst[INET6_ADDRSTRLEN];
- inet_ntop(af, a->src_ip_addr, (void *)src, sizeof(src));
- inet_ntop(af, a->dst_ip_addr, (void *)dst, sizeof(dst));
+ inet_ntop(af, &a->src_prefix.address.un, (void *)src, sizeof(src));
+ inet_ntop(af, &a->dst_prefix.address.un, (void *)dst, sizeof(dst));
out = format(out, "%s action %d src %s/%d dst %s/%d proto %d sport %d-%d dport %d-%d tcpflags %d mask %d",
- a->is_ipv6 ? "ipv6" : "ipv4", a->is_permit,
- src, a->src_ip_prefix_len,
- dst, a->dst_ip_prefix_len,
+ a->src_prefix.address.af ? "ipv6" : "ipv4", a->is_permit,
+ src, a->src_prefix.len,
+ dst, a->dst_prefix.len,
a->proto,
a->srcport_or_icmptype_first, a->srcport_or_icmptype_last,
a->dstport_or_icmpcode_first, a->dstport_or_icmpcode_last,
static inline u8 *
vl_api_macip_acl_rule_t_pretty_format (u8 *out, vl_api_macip_acl_rule_t * a)
{
- int af = a->is_ipv6 ? AF_INET6 : AF_INET;
+ int af = a->src_prefix.address.af ? AF_INET6 : AF_INET;
u8 src[INET6_ADDRSTRLEN];
- inet_ntop(af, a->src_ip_addr, (void *)src, sizeof(src));
+ inet_ntop(af, &a->src_prefix.address.un, (void *)src, sizeof(src));
out = format(out, "%s action %d ip %s/%d mac %U mask %U",
- a->is_ipv6 ? "ipv6" : "ipv4", a->is_permit,
- src, a->src_ip_prefix_len,
+ a->src_prefix.address.af ? "ipv6" : "ipv4", a->is_permit,
+ src, a->src_prefix.len,
my_format_mac_address, a->src_mac,
my_format_mac_address, a->src_mac_mask);
return(out);
while (unformat_check_input (i) != UNFORMAT_END_OF_INPUT)
{
- if (unformat (i, "ipv6"))
- {
- vec_validate_acl_rules(rules, rule_idx);
- rules[rule_idx].is_ipv6 = 1;
- }
- else if (unformat (i, "ipv4"))
- {
- vec_validate_acl_rules(rules, rule_idx);
- rules[rule_idx].is_ipv6 = 0;
- }
- else if (unformat (i, "permit+reflect"))
+ if (unformat (i, "permit+reflect"))
{
vec_validate_acl_rules(rules, rule_idx);
rules[rule_idx].is_permit = 2;
unformat_ip4_address, &src_v4address, &src_prefix_length))
{
vec_validate_acl_rules(rules, rule_idx);
- memcpy (rules[rule_idx].src_ip_addr, &src_v4address, 4);
- rules[rule_idx].src_ip_prefix_len = src_prefix_length;
- rules[rule_idx].is_ipv6 = 0;
+ memcpy (rules[rule_idx].src_prefix.address.un.ip4, &src_v4address, 4);
+ rules[rule_idx].src_prefix.address.af = ADDRESS_IP4;
+ rules[rule_idx].src_prefix.len = src_prefix_length;
}
else if (unformat (i, "src %U/%d",
unformat_ip6_address, &src_v6address, &src_prefix_length))
{
vec_validate_acl_rules(rules, rule_idx);
- memcpy (rules[rule_idx].src_ip_addr, &src_v6address, 16);
- rules[rule_idx].src_ip_prefix_len = src_prefix_length;
- rules[rule_idx].is_ipv6 = 1;
+ memcpy (rules[rule_idx].src_prefix.address.un.ip6, &src_v6address, 16);
+ rules[rule_idx].src_prefix.address.af = ADDRESS_IP6;
+ rules[rule_idx].src_prefix.len = src_prefix_length;
}
else if (unformat (i, "dst %U/%d",
unformat_ip4_address, &dst_v4address, &dst_prefix_length))
{
vec_validate_acl_rules(rules, rule_idx);
- memcpy (rules[rule_idx].dst_ip_addr, &dst_v4address, 4);
- rules[rule_idx].dst_ip_prefix_len = dst_prefix_length;
- rules[rule_idx].is_ipv6 = 0;
+ memcpy (rules[rule_idx].dst_prefix.address.un.ip4, &dst_v4address, 4);
+ rules[rule_idx].dst_prefix.address.af = ADDRESS_IP4;
+ rules[rule_idx].dst_prefix.len = dst_prefix_length;
}
else if (unformat (i, "dst %U/%d",
unformat_ip6_address, &dst_v6address, &dst_prefix_length))
{
vec_validate_acl_rules(rules, rule_idx);
- memcpy (rules[rule_idx].dst_ip_addr, &dst_v6address, 16);
- rules[rule_idx].dst_ip_prefix_len = dst_prefix_length;
- rules[rule_idx].is_ipv6 = 1;
+ memcpy (rules[rule_idx].dst_prefix.address.un.ip6, &dst_v6address, 16);
+ rules[rule_idx].dst_prefix.address.af = ADDRESS_IP6;
+ rules[rule_idx].dst_prefix.len = dst_prefix_length;
}
else if (unformat (i, "sport %d-%d", &port1, &port2))
{
rule_idx++;
vec_validate_acl_rules(rules, rule_idx);
- rules[rule_idx].is_ipv6 = 0;
rules[rule_idx].is_permit = is_permit;
- memcpy (rules[rule_idx].src_ip_addr, &src_v4address, 4);
- rules[rule_idx].src_ip_prefix_len = src_prefix_length;
- memcpy (rules[rule_idx].dst_ip_addr, &dst_v4address, 4);
- rules[rule_idx].dst_ip_prefix_len = dst_prefix_length;
+ memcpy (rules[rule_idx].src_prefix.address.un.ip4, &src_v4address, 4);
+ rules[rule_idx].src_prefix.address.af = ADDRESS_IP4;
+ rules[rule_idx].src_prefix.len = src_prefix_length;
+ memcpy (rules[rule_idx].dst_prefix.address.un.ip4, &dst_v4address, 4);
+ rules[rule_idx].dst_prefix.address.af = ADDRESS_IP4;
+ rules[rule_idx].dst_prefix.len = dst_prefix_length;
rules[rule_idx].srcport_or_icmptype_first = htons(sport_low);
rules[rule_idx].srcport_or_icmptype_last = htons(sport_high);
rules[rule_idx].dstport_or_icmpcode_first = htons(dport_low);
rule_idx++;
vec_validate_acl_rules(rules, rule_idx);
- rules[rule_idx].is_ipv6 = 0;
rules[rule_idx].is_permit = is_permit == 2 ? 2 : 1;
src_v4address.data[0]=0;
src_v4address.data[1]=0;
src_v4address.data[2]=0;
src_v4address.data[3]=0;
- memcpy (rules[rule_idx].src_ip_addr, &src_v4address, 4);
- rules[rule_idx].src_ip_prefix_len = 0;
+ memcpy (rules[rule_idx].src_prefix.address.un.ip4, &src_v4address, 4);
+ rules[rule_idx].src_prefix.address.af = ADDRESS_IP4;
+ rules[rule_idx].src_prefix.len = 0;
dst_v4address.data[0]=0;
dst_v4address.data[1]=0;
dst_v4address.data[2]=0;
dst_v4address.data[3]=0;
- memcpy (rules[rule_idx].dst_ip_addr, &dst_v4address, 4);
- rules[rule_idx].dst_ip_prefix_len = 0;
+ memcpy (rules[rule_idx].dst_prefix.address.un.ip4, &dst_v4address, 4);
+ rules[rule_idx].dst_prefix.address.af = ADDRESS_IP4;
+ rules[rule_idx].dst_prefix.len = 0;
rules[rule_idx].srcport_or_icmptype_first = htons(0);
rules[rule_idx].srcport_or_icmptype_last = htons(65535);
while (unformat_check_input (i) != UNFORMAT_END_OF_INPUT)
{
- if (unformat (i, "ipv6"))
- {
- vec_validate_macip_acl_rules(rules, rule_idx);
- rules[rule_idx].is_ipv6 = 1;
- }
- else if (unformat (i, "ipv4"))
- {
- vec_validate_macip_acl_rules(rules, rule_idx);
- rules[rule_idx].is_ipv6 = 0;
- }
- else if (unformat (i, "permit"))
+ if (unformat (i, "permit"))
{
vec_validate_macip_acl_rules(rules, rule_idx);
rules[rule_idx].is_permit = 1;
if (src_prefix_length == 0)
src_prefix_length = 32;
vec_validate_macip_acl_rules(rules, rule_idx);
- memcpy (rules[rule_idx].src_ip_addr, &src_v4address, 4);
- rules[rule_idx].src_ip_prefix_len = src_prefix_length;
- rules[rule_idx].is_ipv6 = 0;
+ memcpy (rules[rule_idx].src_prefix.address.un.ip4, &src_v4address, 4);
+ rules[rule_idx].src_prefix.address.af = ADDRESS_IP4;
+ rules[rule_idx].src_prefix.len = src_prefix_length;
}
else if (unformat (i, "src"))
{
if (src_prefix_length == 0)
src_prefix_length = 128;
vec_validate_macip_acl_rules(rules, rule_idx);
- memcpy (rules[rule_idx].src_ip_addr, &src_v6address, 16);
- rules[rule_idx].src_ip_prefix_len = src_prefix_length;
- rules[rule_idx].is_ipv6 = 1;
+ memcpy (rules[rule_idx].src_prefix.address.un.ip4, &src_v6address, 4);
+ rules[rule_idx].src_prefix.address.af = ADDRESS_IP6;
+ rules[rule_idx].src_prefix.len = src_prefix_length;
}
else if (unformat (i, "mac %U",
my_unformat_mac_address, &src_mac))
while (unformat_check_input (i) != UNFORMAT_END_OF_INPUT)
{
- if (unformat (i, "ipv6"))
- {
- vec_validate_macip_acl_rules(rules, rule_idx);
- rules[rule_idx].is_ipv6 = 1;
- }
- else if (unformat (i, "ipv4"))
- {
- vec_validate_macip_acl_rules(rules, rule_idx);
- rules[rule_idx].is_ipv6 = 0;
- }
- else if (unformat (i, "permit"))
+ if (unformat (i, "permit"))
{
vec_validate_macip_acl_rules(rules, rule_idx);
rules[rule_idx].is_permit = 1;
rules[rule_idx].is_permit = action;
}
else if (unformat (i, "ip %U/%d",
- unformat_ip4_address, &src_v4address, &src_prefix_length) ||
- unformat (i, "ip %U",
- unformat_ip4_address, &src_v4address))
+ unformat_ip4_address, &src_v4address, &src_prefix_length) ||
+ unformat (i, "ip %U",
+ unformat_ip4_address, &src_v4address))
{
- if (src_prefix_length == 0)
- src_prefix_length = 32;
- vec_validate_macip_acl_rules(rules, rule_idx);
- memcpy (rules[rule_idx].src_ip_addr, &src_v4address, 4);
- rules[rule_idx].src_ip_prefix_len = src_prefix_length;
- rules[rule_idx].is_ipv6 = 0;
+ if (src_prefix_length == 0)
+ src_prefix_length = 32;
+ vec_validate_macip_acl_rules(rules, rule_idx);
+ memcpy (rules[rule_idx].src_prefix.address.un.ip4, &src_v4address, 4);
+ rules[rule_idx].src_prefix.address.af = ADDRESS_IP4;
+ rules[rule_idx].src_prefix.len = src_prefix_length;
}
else if (unformat (i, "src"))
{
- /* Everything in MACIP is "source" but allow this verbosity */
+ /* Everything in MACIP is "source" but allow this verbosity */
}
else if (unformat (i, "ip %U/%d",
- unformat_ip6_address, &src_v6address, &src_prefix_length) ||
- unformat (i, "ip %U",
- unformat_ip6_address, &src_v6address))
+ unformat_ip6_address, &src_v6address, &src_prefix_length) ||
+ unformat (i, "ip %U",
+ unformat_ip6_address, &src_v6address))
{
if (src_prefix_length == 0)
- src_prefix_length = 128;
+ src_prefix_length = 128;
vec_validate_macip_acl_rules(rules, rule_idx);
- memcpy (rules[rule_idx].src_ip_addr, &src_v6address, 16);
- rules[rule_idx].src_ip_prefix_len = src_prefix_length;
- rules[rule_idx].is_ipv6 = 1;
+ memcpy (rules[rule_idx].src_prefix.address.un.ip4, &src_v6address, 4);
+ rules[rule_idx].src_prefix.address.af = ADDRESS_IP6;
+ rules[rule_idx].src_prefix.len = src_prefix_length;
}
else if (unformat (i, "mac %U",
my_unformat_mac_address, &src_mac))
* limitations under the License.
*/
-
+import "vnet/ip/ip_types.api";
+import "vnet/ethernet/ethernet_types.api";
+
+enum acl_action : u8
+{
+ ACL_ACTION_API_DENY = 0,
+ ACL_ACTION_API_PERMIT = 1,
+ ACL_ACTION_API_PERMIT_REFLECT = 2,
+};
+
/** \brief Access List Rule entry
@param is_permit - deny (0), permit (1), or permit+reflect(2) action on this rule.
- @param is_ipv6 - IP addresses in this rule are IPv6 (1) or IPv4 (0)
- @param src_ip_addr - Source prefix value
- @param src_ip_prefix_len - Source prefix length
- @param dst_ip_addr - Destination prefix value
- @param dst_ip_prefix_len - Destination prefix length
+ @param src_prefix - Source prefix
+ @param dst_prefix - Destination prefix
@param proto - L4 protocol (http://www.iana.org/assignments/protocol-numbers/protocol-numbers.xhtml)
@param srcport_or_icmptype_first - beginning of source port or ICMP4/6 type range
@param srcport_or_icmptype_last - end of source port or ICMP4/6 type range
typedef acl_rule
{
- u8 is_permit;
- u8 is_ipv6;
- u8 src_ip_addr[16];
- u8 src_ip_prefix_len;
- u8 dst_ip_addr[16];
- u8 dst_ip_prefix_len;
+ vl_api_acl_action_t is_permit;
+ vl_api_prefix_t src_prefix;
+ vl_api_prefix_t dst_prefix;
/*
* L4 protocol. IANA number. 1 = ICMP, 58 = ICMPv6, 6 = TCP, 17 = UDP.
* 0 => ignore L4 and ignore the ports/tcpflags when matching.
*/
- u8 proto;
+ vl_api_ip_proto_t proto;
/*
* If the L4 protocol is TCP or UDP, the below
* hold ranges of ports, else if the L4 is ICMP/ICMPv6
/** \brief MACIP Access List Rule entry
@param is_permit - deny (0), permit (1) action on this rule.
- @param is_ipv6 - IP addresses in this rule are IPv6 (1) or IPv4 (0)
@param src_mac - match masked source MAC address against this value
@param src_mac_mask - AND source MAC address with this value before matching
- @param src_ip_addr - Source prefix value
- @param src_ip_prefix_len - Source prefix length
+ @param src_prefix - Source prefix value
*/
typedef macip_acl_rule
{
- u8 is_permit;
- u8 is_ipv6;
+ vl_api_acl_action_t is_permit;
/*
* The source mac of the packet ANDed with src_mac_mask.
* The source ip[46] address in the packet is matched
- * against src_ip_addr, with src_ip_prefix_len set to 0.
+ * against src_prefix set to 0.
*
* For better performance, minimize the number of
- * (src_mac_mask, src_ip_prefix_len) combinations
+ * (src_mac_mask, src_prefix.len) combinations
* in a MACIP ACL.
*/
- u8 src_mac[6];
- u8 src_mac_mask[6];
- u8 src_ip_addr[16];
- u8 src_ip_prefix_len;
+ vl_api_mac_address_t src_mac;
+ vl_api_mac_address_t src_mac_mask;
+ vl_api_prefix_t src_prefix;
};
-
-
#include <vnet/ip/format.h>
#include <vnet/ethernet/ethernet.h>
+#include <vnet/ip/ip_types_api.h>
#define vl_endianfun /* define message structures */
#include <acl/acl_types.api.h>
vl_api_acl_rule_t_print (vl_api_acl_rule_t * a, void *handle)
{
u8 *s;
+ fib_prefix_t src, dst;
- s = format (0, " %s ", a->is_ipv6 ? "ipv6" : "ipv4");
+ ip_prefix_decode (&a->src_prefix, &src);
+ ip_prefix_decode (&a->dst_prefix, &dst);
+
+ s = format (0, " %s ", a->src_prefix.address.af ? "ipv6" : "ipv4");
s = format_acl_action (s, a->is_permit);
s = format (s, " \\\n");
- if (a->is_ipv6)
- s = format (s, " src %U/%d dst %U/%d \\\n",
- format_ip6_address, a->src_ip_addr, a->src_ip_prefix_len,
- format_ip6_address, a->dst_ip_addr, a->dst_ip_prefix_len);
- else
- s = format (s, " src %U/%d dst %U/%d \\\n",
- format_ip4_address, a->src_ip_addr, a->src_ip_prefix_len,
- format_ip4_address, a->dst_ip_addr, a->dst_ip_prefix_len);
+ s = format (s, " src %U dst %U \\\n",
+ format_fib_prefix, &src,
+ format_fib_prefix, &dst);
s = format (s, " proto %d \\\n", a->proto);
s = format (s, " sport %d-%d dport %d-%d \\\n",
clib_net_to_host_u16 (a->srcport_or_icmptype_first),
vl_api_macip_acl_rule_t_print (vl_api_macip_acl_rule_t * a, void *handle)
{
u8 *s;
+ fib_prefix_t src;
+
+ ip_prefix_decode (&a->src_prefix, &src);
- s = format (0, " %s %s \\\n", a->is_ipv6 ? "ipv6" : "ipv4",
+ s = format (0, " %s %s \\\n", a->src_prefix.address.af ? "ipv6" : "ipv4",
a->is_permit ? "permit" : "deny");
s = format (s, " src mac %U mask %U \\\n",
format_ethernet_address, a->src_mac,
format_ethernet_address, a->src_mac_mask);
- if (a->is_ipv6)
- s = format (s, " src ip %U/%d, \\",
- format_ip6_address, a->src_ip_addr, a->src_ip_prefix_len);
- else
- s = format (s, " src ip %U/%d, \\",
- format_ip4_address, a->src_ip_addr, a->src_ip_prefix_len);
+ s = format (s, " src ip %U, \\",
+ format_fib_prefix, &src);
PRINT_S;
return handle;
from scapy.layers.inet6 import IPv6ExtHdrFragment
from framework import VppTestCase, VppTestRunner
from util import Host, ppp
+from ipaddress import IPv4Network, IPv6Network
from vpp_lo_interface import VppLoInterface
+from vpp_acl import AclRule, VppAcl, VppAclInterface, VppEtypeWhitelist
+from vpp_ip import INVALID_INDEX
class TestACLplugin(VppTestCase):
% self.bd_id))
def create_rule(self, ip=0, permit_deny=0, ports=PORTS_ALL, proto=-1,
- s_prefix=0, s_ip=b'\x00\x00\x00\x00',
- d_prefix=0, d_ip=b'\x00\x00\x00\x00'):
- if proto == -1:
- return
- if ports == self.PORTS_ALL:
- sport_from = 0
- dport_from = 0
- sport_to = 65535 if proto != 1 and proto != 58 else 255
- dport_to = sport_to
- elif ports == self.PORTS_RANGE:
- if proto == 1:
- sport_from = self.icmp4_type
- sport_to = self.icmp4_type
- dport_from = self.icmp4_code
- dport_to = self.icmp4_code
- elif proto == 58:
- sport_from = self.icmp6_type
- sport_to = self.icmp6_type
- dport_from = self.icmp6_code
- dport_to = self.icmp6_code
- elif proto == self.proto[self.IP][self.TCP]:
- sport_from = self.tcp_sport_from
- sport_to = self.tcp_sport_to
- dport_from = self.tcp_dport_from
- dport_to = self.tcp_dport_to
- elif proto == self.proto[self.IP][self.UDP]:
- sport_from = self.udp_sport_from
- sport_to = self.udp_sport_to
- dport_from = self.udp_dport_from
- dport_to = self.udp_dport_to
- elif ports == self.PORTS_RANGE_2:
- if proto == 1:
- sport_from = self.icmp4_type_2
- sport_to = self.icmp4_type_2
- dport_from = self.icmp4_code_from_2
- dport_to = self.icmp4_code_to_2
- elif proto == 58:
- sport_from = self.icmp6_type_2
- sport_to = self.icmp6_type_2
- dport_from = self.icmp6_code_from_2
- dport_to = self.icmp6_code_to_2
- elif proto == self.proto[self.IP][self.TCP]:
- sport_from = self.tcp_sport_from_2
- sport_to = self.tcp_sport_to_2
- dport_from = self.tcp_dport_from_2
- dport_to = self.tcp_dport_to_2
- elif proto == self.proto[self.IP][self.UDP]:
- sport_from = self.udp_sport_from_2
- sport_to = self.udp_sport_to_2
- dport_from = self.udp_dport_from_2
- dport_to = self.udp_dport_to_2
+ s_prefix=0, s_ip=0,
+ d_prefix=0, d_ip=0):
+ if ip:
+ src_prefix = IPv6Network((s_ip, s_prefix))
+ dst_prefix = IPv6Network((d_ip, d_prefix))
else:
- sport_from = ports
- sport_to = ports
- dport_from = ports
- dport_to = ports
-
- rule = ({'is_permit': permit_deny, 'is_ipv6': ip, 'proto': proto,
- 'srcport_or_icmptype_first': sport_from,
- 'srcport_or_icmptype_last': sport_to,
- 'src_ip_prefix_len': s_prefix,
- 'src_ip_addr': s_ip,
- 'dstport_or_icmpcode_first': dport_from,
- 'dstport_or_icmpcode_last': dport_to,
- 'dst_ip_prefix_len': d_prefix,
- 'dst_ip_addr': d_ip})
- return rule
-
- def apply_rules(self, rules, tag=b''):
- reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules,
- tag=tag)
- self.logger.info("Dumped ACL: " + str(
- self.vapi.acl_dump(reply.acl_index)))
+ src_prefix = IPv4Network((s_ip, s_prefix))
+ dst_prefix = IPv4Network((d_ip, d_prefix))
+ return AclRule(is_permit=permit_deny, ports=ports, proto=proto,
+ src_prefix=src_prefix, dst_prefix=dst_prefix)
+
+ def apply_rules(self, rules, tag=None):
+ acl = VppAcl(self, rules, tag=tag)
+ acl.add_vpp_config()
+ self.logger.info("Dumped ACL: " + str(acl.dump()))
# Apply a ACL on the interface as inbound
for i in self.pg_interfaces:
- self.vapi.acl_interface_set_acl_list(sw_if_index=i.sw_if_index,
- n_input=1,
- acls=[reply.acl_index])
- return reply.acl_index
-
- def apply_rules_to(self, rules, tag=b'', sw_if_index=0xFFFFFFFF):
- reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules,
- tag=tag)
- self.logger.info("Dumped ACL: " + str(
- self.vapi.acl_dump(reply.acl_index)))
+ acl_if = VppAclInterface(
+ self, sw_if_index=i.sw_if_index, n_input=1, acls=[acl])
+ acl_if.add_vpp_config()
+ return acl.acl_index
+
+ def apply_rules_to(self, rules, tag=None, sw_if_index=INVALID_INDEX):
+ acl = VppAcl(self, rules, tag=tag)
+ acl.add_vpp_config()
+ self.logger.info("Dumped ACL: " + str(acl.dump()))
# Apply a ACL on the interface as inbound
- self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index,
- n_input=1,
- acls=[reply.acl_index])
- return reply.acl_index
+ acl_if = VppAclInterface(self, sw_if_index=sw_if_index, n_input=1,
+ acls=[acl])
+ return acl.acl_index
- def etype_whitelist(self, whitelist, n_input):
+ def etype_whitelist(self, whitelist, n_input, add=True):
# Apply whitelists on all the interfaces
- for i in self.pg_interfaces:
- # checkstyle can't read long names. Help them.
- fun = self.vapi.acl_interface_set_etype_whitelist
- fun(sw_if_index=i.sw_if_index, n_input=n_input,
- whitelist=whitelist)
- return
+ if add:
+ self._wl = []
+ for i in self.pg_interfaces:
+ self._wl.append(VppEtypeWhitelist(
+ self, sw_if_index=i.sw_if_index, whitelist=whitelist,
+ n_input=n_input).add_vpp_config())
+ else:
+ if hasattr(self, "_wl"):
+ for wl in self._wl:
+ wl.remove_vpp_config()
def create_upper_layer(self, packet_index, proto, ports=0):
p = self.proto_map[proto]
"""
self.logger.info("ACLP_TEST_START_0001")
- # Add an ACL
- r = [{'is_permit': 1, 'is_ipv6': 0, 'proto': 17,
- 'srcport_or_icmptype_first': 1234,
- 'srcport_or_icmptype_last': 1235,
- 'src_ip_prefix_len': 0,
- 'src_ip_addr': b'\x00\x00\x00\x00',
- 'dstport_or_icmpcode_first': 1234,
- 'dstport_or_icmpcode_last': 1234,
- 'dst_ip_addr': b'\x00\x00\x00\x00',
- 'dst_ip_prefix_len': 0}]
+ # Create a permit-1234 ACL
+ r = [AclRule(is_permit=1, proto=17, ports=1234)]
+ r[0].sport_to = 1235
# Test 1: add a new ACL
- reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r,
- tag=b"permit 1234")
- self.assertEqual(reply.retval, 0)
+ first_acl = VppAcl(self, rules=r, tag="permit 1234")
+ first_acl.add_vpp_config()
+ self.assertTrue(first_acl.query_vpp_config())
# The very first ACL gets #0
- self.assertEqual(reply.acl_index, 0)
- first_acl = reply.acl_index
- rr = self.vapi.acl_dump(reply.acl_index)
+ self.assertEqual(first_acl.acl_index, 0)
+ rr = first_acl.dump()
self.logger.info("Dumped ACL: " + str(rr))
self.assertEqual(len(rr), 1)
# We should have the same number of ACL entries as we had asked
# are different types, we need to iterate over rules and keys to get
# to basic values.
for i_rule in range(0, len(r) - 1):
- for rule_key in r[i_rule]:
+ encoded_rule = r[i_rule].encode()
+ for rule_key in encoded_rule:
self.assertEqual(rr[0].r[i_rule][rule_key],
- r[i_rule][rule_key])
-
- # Add a deny-1234 ACL
- r_deny = [{'is_permit': 0, 'is_ipv6': 0, 'proto': 17,
- 'srcport_or_icmptype_first': 1234,
- 'srcport_or_icmptype_last': 1235,
- 'src_ip_prefix_len': 0,
- 'src_ip_addr': b'\x00\x00\x00\x00',
- 'dstport_or_icmpcode_first': 1234,
- 'dstport_or_icmpcode_last': 1234,
- 'dst_ip_addr': b'\x00\x00\x00\x00',
- 'dst_ip_prefix_len': 0},
- {'is_permit': 1, 'is_ipv6': 0, 'proto': 17,
- 'srcport_or_icmptype_first': 0,
- 'srcport_or_icmptype_last': 0,
- 'src_ip_prefix_len': 0,
- 'src_ip_addr': b'\x00\x00\x00\x00',
- 'dstport_or_icmpcode_first': 0,
- 'dstport_or_icmpcode_last': 0,
- 'dst_ip_addr': b'\x00\x00\x00\x00',
- 'dst_ip_prefix_len': 0}]
-
- reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_deny,
- tag=b"deny 1234;permit all")
- self.assertEqual(reply.retval, 0)
+ encoded_rule[rule_key])
+
+ # Create a deny-1234 ACL
+ r_deny = [AclRule(is_permit=0, proto=17, ports=1234),
+ AclRule(is_permit=1, proto=17, ports=0)]
+ r_deny[0].sport_to = 1235
+ second_acl = VppAcl(self, rules=r_deny, tag="deny 1234;permit all")
+ second_acl.add_vpp_config()
+ self.assertTrue(second_acl.query_vpp_config())
# The second ACL gets #1
- self.assertEqual(reply.acl_index, 1)
- second_acl = reply.acl_index
+ self.assertEqual(second_acl.acl_index, 1)
# Test 2: try to modify a nonexistent ACL
- reply = self.vapi.acl_add_replace(acl_index=432, r=r,
- tag=b"FFFF:FFFF", expected_retval=-6)
- self.assertEqual(reply.retval, -6)
- # The ACL number should pass through
- self.assertEqual(reply.acl_index, 432)
+ invalid_acl = VppAcl(self, acl_index=432, rules=r, tag="FFFF:FFFF")
+ reply = invalid_acl.add_vpp_config(expect_error=True)
+
+ # apply an ACL on an interface inbound, try to delete ACL, must fail
+ acl_if_list = VppAclInterface(
+ self, sw_if_index=self.pg0.sw_if_index, n_input=1,
+ acls=[first_acl])
+ acl_if_list.add_vpp_config()
+ first_acl.remove_vpp_config(expect_error=True)
+ # Unapply an ACL and then try to delete it - must be ok
+ acl_if_list.remove_vpp_config()
+ first_acl.remove_vpp_config()
+
# apply an ACL on an interface inbound, try to delete ACL, must fail
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
- n_input=1,
- acls=[first_acl])
- reply = self.vapi.acl_del(acl_index=first_acl, expected_retval=-142)
+ acl_if_list = VppAclInterface(
+ self, sw_if_index=self.pg0.sw_if_index, n_input=0,
+ acls=[second_acl])
+ acl_if_list.add_vpp_config()
+ second_acl.remove_vpp_config(expect_error=True)
# Unapply an ACL and then try to delete it - must be ok
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
- n_input=0,
- acls=[])
- reply = self.vapi.acl_del(acl_index=first_acl, expected_retval=0)
-
- # apply an ACL on an interface outbound, try to delete ACL, must fail
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
- n_input=0,
- acls=[second_acl])
- reply = self.vapi.acl_del(acl_index=second_acl, expected_retval=-143)
- # Unapply the ACL and then try to delete it - must be ok
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
- n_input=0,
- acls=[])
- reply = self.vapi.acl_del(acl_index=second_acl, expected_retval=0)
+ acl_if_list.remove_vpp_config()
+ second_acl.remove_vpp_config()
# try to apply a nonexistent ACL - must fail
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
- n_input=1,
- acls=[first_acl],
- expected_retval=-6)
+ acl_if_list = VppAclInterface(
+ self, sw_if_index=self.pg0.sw_if_index, n_input=0,
+ acls=[invalid_acl])
+ acl_if_list.add_vpp_config(expect_error=True)
self.logger.info("ACLP_TEST_FINISH_0001")
rules = []
rules.append(self.create_rule(self.IPV4, self.PERMIT,
- 0, self.proto[self.IP][self.UDP]))
+ 0, self.proto[self.IP][self.UDP]))
rules.append(self.create_rule(self.IPV4, self.PERMIT,
- 0, self.proto[self.IP][self.TCP]))
+ 0, self.proto[self.IP][self.TCP]))
# Apply rules
- acl_idx = self.apply_rules(rules, b"permit per-flow")
+ acl_idx = self.apply_rules(rules, "permit per-flow")
# enable counters
reply = self.vapi.papi.acl_stats_intf_counters_enable(enable=1)
self.logger.info("ACLP_TEST_START_0003")
# Add a deny-flows ACL
rules = []
- rules.append(self.create_rule(self.IPV4, self.DENY,
- self.PORTS_ALL, self.proto[self.IP][self.UDP]))
+ rules.append(self.create_rule(
+ self.IPV4, self.DENY, self.PORTS_ALL,
+ self.proto[self.IP][self.UDP]))
# Permit ip any any in the end
rules.append(self.create_rule(self.IPV4, self.PERMIT,
self.PORTS_ALL, 0))
# Apply rules
- acl_idx = self.apply_rules(rules, b"deny per-flow;permit all")
+ acl_idx = self.apply_rules(rules, "deny per-flow;permit all")
# enable counters
reply = self.vapi.papi.acl_stats_intf_counters_enable(enable=1)
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit icmpv4")
+ self.apply_rules(rules, "permit icmpv4")
# Traffic should still pass
self.run_verify_test(self.ICMP, self.IPV4,
rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit icmpv6")
+ self.apply_rules(rules, "permit icmpv6")
# Traffic should still pass
self.run_verify_test(self.ICMP, self.IPV6,
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"deny icmpv4")
+ self.apply_rules(rules, "deny icmpv4")
# Traffic should not pass
self.run_verify_negat_test(self.ICMP, self.IPV4, 0)
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"deny icmpv6")
+ self.apply_rules(rules, "deny icmpv6")
# Traffic should not pass
self.run_verify_negat_test(self.ICMP, self.IPV6, 0)
# Add an ACL
rules = []
rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
# deny ip any any in the end
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ipv4 tcp")
+ self.apply_rules(rules, "permit ipv4 tcp")
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP])
rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ip6 tcp")
+ self.apply_rules(rules, "permit ip6 tcp")
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.TCP])
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ipv udp")
+ self.apply_rules(rules, "permit ipv udp")
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.UDP])
rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ip6 udp")
+ self.apply_rules(rules, "permit ip6 udp")
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.UDP])
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"deny ip4/ip6 tcp")
+ self.apply_rules(rules, "deny ip4/ip6 tcp")
# Traffic should not pass
self.run_verify_negat_test(self.IP, self.IPRANDOM,
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"deny ip4/ip6 udp")
+ self.apply_rules(rules, "deny ip4/ip6 udp")
# Traffic should not pass
self.run_verify_negat_test(self.IP, self.IPRANDOM,
for i in range(len(r)):
rules.append(self.create_rule(r[i][0], r[i][1], r[i][2], r[i][3]))
- reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules)
- result = self.vapi.acl_dump(reply.acl_index)
+ acl = VppAcl(self, rules=rules)
+ acl.add_vpp_config()
+ result = acl.dump()
i = 0
for drules in result:
for dr in drules.r:
- self.assertEqual(dr.is_ipv6, r[i][0])
self.assertEqual(dr.is_permit, r[i][1])
self.assertEqual(dr.proto, r[i][3])
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ip4 tcp %d" % port)
+ self.apply_rules(rules, "permit ip4 tcp %d" % port)
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV4,
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ip4 tcp %d" % port)
+ self.apply_rules(rules, "permit ip4 tcp %d" % port)
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV4,
rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ip4 tcp %d" % port)
+ self.apply_rules(rules, "permit ip4 tcp %d" % port)
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV6,
self.logger.info("ACLP_TEST_FINISH_0017")
def test_0018_udp_permit_port_v6(self):
- """ permit single UPPv6
+ """ permit single UDPv6
"""
self.logger.info("ACLP_TEST_START_0018")
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ip4 tcp %d" % port)
+ self.apply_rules(rules, "permit ip4 tcp %d" % port)
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV6,
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"deny ip4/ip6 udp %d" % port)
+ self.apply_rules(rules, "deny ip4/ip6 udp %d" % port)
# Traffic should not pass
self.run_verify_negat_test(self.IP, self.IPRANDOM,
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"deny ip4/ip6 udp %d" % port)
+ self.apply_rules(rules, "deny ip4/ip6 udp %d" % port)
# Traffic should not pass
self.run_verify_negat_test(self.IP, self.IPRANDOM,
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"deny ip4/ip6 udp %d" % port)
+ self.apply_rules(rules, "deny ip4/ip6 udp %d" % port)
# Traffic should not pass
self.run_verify_negat_test(self.IP, self.IPRANDOM,
self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit empty udp ip4 %d" % port)
+ self.apply_rules(rules, "permit empty udp ip4 %d" % port)
# Traffic should still pass
# Create incoming packet streams for packet-generator interfaces
rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit empty udp ip6 %d" % port)
+ self.apply_rules(rules, "permit empty udp ip6 %d" % port)
# Traffic should still pass
# Create incoming packet streams for packet-generator interfaces
# Add an ACL
rules = []
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
# deny ip any any in the end
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ipv4 tcp")
+ self.apply_rules(rules, "permit ipv4 tcp")
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP])
rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ip6 tcp")
+ self.apply_rules(rules, "permit ip6 tcp")
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.TCP])
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ipv4 udp")
+ self.apply_rules(rules, "permit ipv4 udp")
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.UDP])
rules.append(self.create_rule(self.IPV6, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ip6 udp")
+ self.apply_rules(rules, "permit ip6 udp")
# Traffic should still pass
self.run_verify_test(self.IP, self.IPV6, self.proto[self.IP][self.UDP])
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"deny ip4/ip6 tcp")
+ self.apply_rules(rules, "deny ip4/ip6 tcp")
# Traffic should not pass
self.run_verify_negat_test(self.IP, self.IPRANDOM,
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"deny ip4/ip6 udp")
+ self.apply_rules(rules, "deny ip4/ip6 udp")
# Traffic should not pass
self.run_verify_negat_test(self.IP, self.IPRANDOM,
# Add an ACL
rules = []
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
# deny ip any any in the end
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ipv4 tcp")
+ self.apply_rules(rules, "permit ipv4 tcp")
# Traffic should still pass also for an odd ethertype
self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP],
# Add an ACL
rules = []
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
# deny ip any any in the end
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ipv4 tcp")
-
+ self.apply_rules(rules, "permit ipv4 tcp")
# whitelist the 0xbbbb etype - so the 0xaaaa should be blocked
self.etype_whitelist([0xbbb], 1)
0, False, 0xaaaa)
# remove the whitelist
- self.etype_whitelist([], 0)
+ self.etype_whitelist([], 0, add=False)
self.logger.info("ACLP_TEST_FINISH_0305")
# Add an ACL
rules = []
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
# deny ip any any in the end
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ipv4 tcp")
-
+ self.apply_rules(rules, "permit ipv4 tcp")
# whitelist the 0xbbbb etype - so the 0xaaaa should be blocked
self.etype_whitelist([0xbbb], 1)
0, False, True, 0x0bbb)
# remove the whitelist, the previously blocked 0xAAAA should pass now
- self.etype_whitelist([], 0)
+ self.etype_whitelist([], 0, add=False)
self.logger.info("ACLP_TEST_FINISH_0306")
# Add an ACL
rules = []
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
# deny ip any any in the end
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, b"permit ipv4 tcp")
+ self.apply_rules(rules, "permit ipv4 tcp")
# whitelist the 0xbbbb etype - so the 0xaaaa should be blocked
self.etype_whitelist([0xbbb], 1)
# remove the whitelist, the previously blocked 0xAAAA should pass now
- self.etype_whitelist([], 0)
+ self.etype_whitelist([], 0, add=False)
# The whitelisted traffic, should pass
self.run_verify_test(self.IP, self.IPV4, self.proto[self.IP][self.TCP],
# Add an ACL
rules = []
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_RANGE_2,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
rules.append(self.create_rule(self.IPV4, self.PERMIT, self.PORTS_RANGE,
- self.proto[self.IP][self.TCP]))
+ self.proto[self.IP][self.TCP]))
# deny ip any any in the end
rules.append(self.create_rule(self.IPV4, self.DENY, self.PORTS_ALL, 0))
intf.append(VppLoInterface(self))
# Apply rules
- self.apply_rules_to(rules, b"permit ipv4 tcp", intf[0].sw_if_index)
+ self.apply_rules_to(rules, "permit ipv4 tcp", intf[0].sw_if_index)
# Remove the interface
intf[0].remove_vpp_config()
self.logger.info("ACLP_TEST_FINISH_0315")
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
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,
- }
+ '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
if reflect_side == acl_side:
self.testcase.vapi.acl_interface_set_acl_list(
- self.ifs[acl_side].sw_if_index, 1,
- [reflect_acl_index,
+ 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, [])
+ 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,
+ 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, [])
+ 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,
- }
+ '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
"""
+import copy
import unittest
from socket import inet_pton, AF_INET, AF_INET6
from random import choice, shuffle
from pprint import pprint
+from ipaddress import ip_network
import scapy.compat
from scapy.packet import Raw
from vpp_l2 import L2_PORT_TYPE
import time
+from vpp_acl import AclRule, VppAcl, VppAclInterface
+
class TestACLpluginL2L3(VppTestCase):
"""TestACLpluginL2L3 Test Case"""
else:
rule_l4_proto = p[IP].proto
- 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,
- 'srcport_or_icmptype_last': rule_l4_sport,
- 'dstport_or_icmpcode_first': rule_l4_dport,
- 'dstport_or_icmpcode_last': rule_l4_dport,
- 'proto': rule_l4_proto,
- }
+ 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)))
+ new_rule.sport_from = rule_l4_sport
+ new_rule.sport_fo = rule_l4_sport
+ new_rule.dport_from = rule_l4_dport
+ new_rule.dport_to = rule_l4_dport
+
rules.append(new_rule)
- new_rule_permit = new_rule.copy()
- new_rule_permit['is_permit'] = 1
+ new_rule_permit = copy.copy(new_rule)
+ new_rule_permit.is_permit = 1
permit_rules.append(new_rule_permit)
- new_rule_permit_and_reflect = new_rule.copy()
+ new_rule_permit_and_reflect = copy.copy(new_rule)
if can_reflect_this_packet:
- new_rule_permit_and_reflect['is_permit'] = 2
+ new_rule_permit_and_reflect.is_permit = 2
else:
- new_rule_permit_and_reflect['is_permit'] = is_permit
+ new_rule_permit_and_reflect.is_permit = is_permit
permit_and_reflect_rules.append(new_rule_permit_and_reflect)
self.logger.info("create_stream pkt#%d: %s" % (i, payload))
# UDP:
- def applied_acl_shuffle(self, sw_if_index):
- # first collect what ACLs are applied and what they look like
- r = self.vapi.acl_interface_list_dump(sw_if_index=sw_if_index)
- orig_applied_acls = r[0]
-
- # we will collect these just to save and generate additional rulesets
- orig_acls = []
- for acl_num in orig_applied_acls.acls:
- rr = self.vapi.acl_dump(acl_num)
- orig_acls.append(rr[0])
+ def applied_acl_shuffle(self, acl_if):
+ saved_n_input = acl_if.n_input
+ # TOTO: maybe copy each one??
+ saved_acls = acl_if.acls
# now create a list of all the rules in all ACLs
all_rules = []
- for old_acl in orig_acls:
- for rule in old_acl.r:
- all_rules.append(dict(rule._asdict()))
+ for old_acl in saved_acls:
+ for rule in old_acl.rules:
+ all_rules.append(rule)
# Add a few ACLs made from shuffled rules
shuffle(all_rules)
- reply = self.vapi.acl_add_replace(acl_index=4294967295,
- r=all_rules[::2],
- tag=b"shuffle 1. acl")
- shuffle_acl_1 = reply.acl_index
+ acl1 = VppAcl(self, rules=all_rules[::2], tag="shuffle 1. acl")
+ acl1.add_vpp_config()
+
shuffle(all_rules)
- reply = self.vapi.acl_add_replace(acl_index=4294967295,
- r=all_rules[::3],
- tag=b"shuffle 2. acl")
- shuffle_acl_2 = reply.acl_index
+ acl2 = VppAcl(self, rules=all_rules[::3], tag="shuffle 2. acl")
+ acl2.add_vpp_config()
+
shuffle(all_rules)
- reply = self.vapi.acl_add_replace(acl_index=4294967295,
- r=all_rules[::2],
- tag=b"shuffle 3. acl")
- shuffle_acl_3 = reply.acl_index
+ acl3 = VppAcl(self, rules=all_rules[::2], tag="shuffle 3. acl")
+ acl3.add_vpp_config()
# apply the shuffle ACLs in front
- input_acls = [shuffle_acl_1, shuffle_acl_2]
- output_acls = [shuffle_acl_1, shuffle_acl_2]
+ input_acls = [acl1, acl2]
+ output_acls = [acl1, acl2]
# add the currently applied ACLs
- n_input = orig_applied_acls.n_input
- input_acls.extend(orig_applied_acls.acls[:n_input])
- output_acls.extend(orig_applied_acls.acls[n_input:])
+ n_input = acl_if.n_input
+ input_acls.extend(saved_acls[:n_input])
+ output_acls.extend(saved_acls[n_input:])
# and the trailing shuffle ACL(s)
- input_acls.extend([shuffle_acl_3])
- output_acls.extend([shuffle_acl_3])
+ input_acls.extend([acl3])
+ output_acls.extend([acl3])
# set the interface ACL list to the result
- self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index,
- n_input=len(input_acls),
- acls=input_acls + output_acls)
+ acl_if.n_input = len(input_acls)
+ acl_if.acls = input_acls + output_acls
+ acl_if.add_vpp_config()
+
# change the ACLs a few times
for i in range(1, 10):
shuffle(all_rules)
- reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_1,
- r=all_rules[::1+(i % 2)],
- tag=b"shuffle 1. acl")
+ acl1.rules = all_rules[::1+(i % 2)]
+ acl1.add_vpp_config()
+
shuffle(all_rules)
- reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_2,
- r=all_rules[::1+(i % 3)],
- tag=b"shuffle 2. acl")
+ acl2.rules = all_rules[::1+(i % 3)]
+ acl2.add_vpp_config()
+
shuffle(all_rules)
- reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_2,
- r=all_rules[::1+(i % 5)],
- tag=b"shuffle 3. acl")
+ acl3.rules = all_rules[::1+(i % 5)]
+ acl3.add_vpp_config()
# restore to how it was before and clean up
- self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index,
- n_input=orig_applied_acls.n_input,
- acls=orig_applied_acls.acls)
- reply = self.vapi.acl_del(acl_index=shuffle_acl_1)
- reply = self.vapi.acl_del(acl_index=shuffle_acl_2)
- reply = self.vapi.acl_del(acl_index=shuffle_acl_3)
+ acl_if.n_input = saved_n_input
+ acl_if.acls = saved_acls
+ acl_if.add_vpp_config()
+
+ acl1.remove_vpp_config()
+ acl2.remove_vpp_config()
+ acl3.remove_vpp_config()
def create_acls_for_a_stream(self, stream_dict,
test_l2_action, is_reflect):
r_permit = stream_dict['permit_rules']
r_permit_reflect = stream_dict['permit_and_reflect_rules']
r_action = r_permit_reflect if is_reflect else r
- reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_action,
- tag=b"act. acl")
- action_acl_index = reply.acl_index
- reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r_permit,
- tag=b"perm. acl")
- permit_acl_index = reply.acl_index
- return {'L2': action_acl_index if test_l2_action else permit_acl_index,
- 'L3': permit_acl_index if test_l2_action else action_acl_index,
- 'permit': permit_acl_index, 'action': action_acl_index}
+ action_acl = VppAcl(self, rules=r_action, tag="act. acl")
+ action_acl.add_vpp_config()
+ permit_acl = VppAcl(self, rules=r_permit, tag="perm. acl")
+ permit_acl.add_vpp_config()
+
+ return {'L2': action_acl if test_l2_action else permit_acl,
+ 'L3': permit_acl if test_l2_action else action_acl,
+ 'permit': permit_acl, 'action': action_acl}
def apply_acl_ip46_x_to_y(self, bridged_to_routed, test_l2_deny,
is_ip6, is_reflect, add_eh):
"""
self.reset_packet_infos()
stream_dict = self.create_stream(
- self.pg2, self.loop0,
- bridged_to_routed,
- self.pg_if_packet_sizes, is_ip6,
- not is_reflect, False, add_eh)
+ self.pg2, self.loop0,
+ bridged_to_routed,
+ self.pg_if_packet_sizes, is_ip6,
+ not is_reflect, False, add_eh)
stream = stream_dict['stream']
acl_idx = self.create_acls_for_a_stream(stream_dict, test_l2_deny,
is_reflect)
n_input_l3 = 0 if bridged_to_routed else 1
n_input_l2 = 1 if bridged_to_routed else 0
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg2.sw_if_index,
- n_input=n_input_l3,
- acls=[acl_idx['L3']])
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
- n_input=n_input_l2,
- acls=[acl_idx['L2']])
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
- n_input=n_input_l2,
- acls=[acl_idx['L2']])
- self.applied_acl_shuffle(self.pg0.sw_if_index)
- self.applied_acl_shuffle(self.pg2.sw_if_index)
+
+ acl_if_pg2 = VppAclInterface(self, sw_if_index=self.pg2.sw_if_index,
+ n_input=n_input_l3, acls=[acl_idx['L3']])
+ acl_if_pg2.add_vpp_config()
+
+ acl_if_pg0 = VppAclInterface(self, sw_if_index=self.pg0.sw_if_index,
+ n_input=n_input_l2, acls=[acl_idx['L2']])
+ acl_if_pg0.add_vpp_config()
+
+ acl_if_pg1 = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index,
+ n_input=n_input_l2, acls=[acl_idx['L2']])
+ acl_if_pg1.add_vpp_config()
+
+ self.applied_acl_shuffle(acl_if_pg0)
+ self.applied_acl_shuffle(acl_if_pg1)
return {'L2': acl_idx['L2'], 'L3': acl_idx['L3']}
def apply_acl_ip46_both_directions_reflect(self,
else:
outbound_l3_acl = acl_idx_rev['L3']
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg2.sw_if_index,
- n_input=1,
- acls=[inbound_l3_acl,
- outbound_l3_acl])
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
- n_input=1,
- acls=[inbound_l2_acl,
- outbound_l2_acl])
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
- n_input=1,
- acls=[inbound_l2_acl,
- outbound_l2_acl])
- self.applied_acl_shuffle(self.pg0.sw_if_index)
- self.applied_acl_shuffle(self.pg2.sw_if_index)
+ acl_if_pg2 = VppAclInterface(self, sw_if_index=self.pg2.sw_if_index,
+ n_input=1,
+ acls=[inbound_l3_acl, outbound_l3_acl])
+ acl_if_pg2.add_vpp_config()
+
+ acl_if_pg0 = VppAclInterface(self, sw_if_index=self.pg0.sw_if_index,
+ n_input=1,
+ acls=[inbound_l2_acl, outbound_l2_acl])
+ acl_if_pg0.add_vpp_config()
+
+ acl_if_pg1 = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index,
+ n_input=1,
+ acls=[inbound_l2_acl, outbound_l2_acl])
+ acl_if_pg1.add_vpp_config()
+
+ self.applied_acl_shuffle(acl_if_pg0)
+ self.applied_acl_shuffle(acl_if_pg2)
def apply_acl_ip46_routed_to_bridged(self, test_l2_deny, is_ip6,
is_reflect, add_eh):
pkts = self.run_traffic_ip46_routed_to_bridged(test_l2_deny, is_ip6,
is_reflect, False,
add_eh)
- self.verify_acl_packet_count(acls['L3'], pkts)
+ self.verify_acl_packet_count(acls['L3'].acl_index, pkts)
def run_test_ip46_bridged_to_routed(self, test_l2_deny,
is_ip6, is_reflect, add_eh):
pkts = self.run_traffic_ip46_bridged_to_routed(test_l2_deny, is_ip6,
is_reflect, False,
add_eh)
- self.verify_acl_packet_count(acls['L2'], pkts)
+ self.verify_acl_packet_count(acls['L2'].acl_index, pkts)
def run_test_ip46_routed_to_bridged_and_back(self, test_l2_action,
is_ip6, add_eh,
from struct import pack, unpack
import re
import unittest
+from ipaddress import ip_network, IPv4Network, IPv6Network
import scapy.compat
from scapy.packet import Raw
from vpp_l2 import L2_PORT_TYPE
from vpp_sub_interface import L2_VTR_OP, VppSubInterface, VppDot1QSubint, \
VppDot1ADSubint
+from vpp_acl import AclRule, VppAcl, VppAclInterface, VppEtypeWhitelist, \
+ VppMacipAclInterface, VppMacipAcl, MacipRule
+from vpp_papi import MACAddress
class MethodHolder(VppTestCase):
# create 2 subinterfaces
cls.subifs = [
- VppDot1QSubint(cls, cls.pg1, 10),
- VppDot1ADSubint(cls, cls.pg2, 20, 300, 400),
- VppDot1QSubint(cls, cls.pg3, 30),
- VppDot1ADSubint(cls, cls.pg3, 40, 600, 700)]
+ VppDot1QSubint(cls, cls.pg1, 10),
+ VppDot1ADSubint(cls, cls.pg2, 20, 300, 400),
+ VppDot1QSubint(cls, cls.pg3, 30),
+ VppDot1ADSubint(cls, cls.pg3, 40, 600, 700)]
cls.subifs[0].set_vtr(L2_VTR_OP.L2_POP_1,
inner=10, push1q=1)
def setUp(self):
super(MethodHolder, self).setUp()
self.reset_packet_infos()
- del self.ACLS[:]
-
- def tearDown(self):
- super(MethodHolder, self).tearDown()
- self.delete_acls()
def show_commands_at_teardown(self):
self.logger.info(self.vapi.ppcli("show interface address"))
acls = self.vapi.macip_acl_dump()
if self.DEBUG:
for acl in acls:
- print("ACL #"+str(acl.acl_index))
+ # print("ACL #"+str(acl.acl_index))
for r in acl.r:
rule = "ACTION"
if r.is_permit == 1:
rule = "PERMIT"
elif r.is_permit == 0:
rule = "DENY "
+ """
print(" IP6" if r.is_ipv6 else " IP4",
rule,
binascii.hexlify(r.src_mac),
binascii.hexlify(r.src_mac_mask),
unpack('<16B', r.src_ip_addr),
r.src_ip_prefix_len)
+ """
return acls
def create_rules(self, mac_type=EXACT_MAC, ip_type=EXACT_IP,
elif ip_type == self.SUBNET_IP:
ip4[2] = random.randint(100, 200)
ip4[3] = 0
- ip6[8] = random.randint(100, 200)
+ ip6[7] = random.randint(100, 200)
ip6[15] = 0
ip_pack = b''
for j in range(0, len(ip)):
ip_pack += pack('<B', int(ip[j]))
- rule = ({'is_permit': self.PERMIT,
- 'is_ipv6': is_ip6,
- 'src_ip_addr': ip_pack,
- 'src_ip_prefix_len': ip_len,
- 'src_mac': binascii.unhexlify(mac.replace(':', '')),
- 'src_mac_mask': binascii.unhexlify(
- mask.replace(':', ''))})
+ rule = MacipRule(is_permit=self.PERMIT,
+ src_prefix=ip_network((ip_pack, ip_len)),
+ src_mac=MACAddress(mac).packed,
+ src_mac_mask=MACAddress(mask).packed)
rules.append(rule)
if ip_type == self.WILD_IP:
break
return acls
def apply_macip_rules(self, acls):
+ macip_acls = []
for acl in acls:
- reply = self.vapi.macip_acl_add(acl)
- self.assertEqual(reply.retval, 0)
- self.ACLS.append(reply.acl_index)
+ macip_acl = VppMacipAcl(self, rules=acl)
+ macip_acl.add_vpp_config()
+ macip_acls.append(macip_acl)
+ return macip_acls
def verify_macip_acls(self, acl_count, rules_count, expected_count=2):
reply = self.macip_acl_dump_debug()
reply = self.vapi.macip_acl_interface_get()
self.assertEqual(reply.count, expected_count)
- def delete_acls(self):
- for acl in range(len(self.ACLS)-1, -1, -1):
- self.vapi.macip_acl_del(self.ACLS[acl])
-
- reply = self.vapi.macip_acl_dump()
- self.assertEqual(len(reply), 0)
-
- intf_acls = self.vapi.acl_interface_list_dump()
- for i_a in intf_acls:
- sw_if_index = i_a.sw_if_index
- for acl_index in i_a.acls:
- self.vapi.acl_interface_add_del(sw_if_index, acl_index, 0)
- self.vapi.acl_del(acl_index)
-
def create_stream(self, mac_type, ip_type, packet_count,
src_if, dst_if, traffic, is_ip6, tags=PERMIT_TAGS):
# exact MAC and exact IP
else:
rule_l4_proto = packet[IP].proto
- acl_rule = {
- 'is_permit': is_permit,
- 'is_ipv6': is_ip6,
- 'src_ip_addr': inet_pton(rule_family,
- packet[rule_l3_layer].src),
- 'src_ip_prefix_len': rule_prefix_len,
- 'dst_ip_addr': inet_pton(rule_family,
- packet[rule_l3_layer].dst),
- 'dst_ip_prefix_len': rule_prefix_len,
- 'srcport_or_icmptype_first': rule_l4_sport,
- 'srcport_or_icmptype_last': rule_l4_sport,
- 'dstport_or_icmpcode_first': rule_l4_dport,
- 'dstport_or_icmpcode_last': rule_l4_dport,
- 'proto': rule_l4_proto}
+ src_network = ip_network(
+ (packet[rule_l3_layer].src, rule_prefix_len))
+ dst_network = ip_network(
+ (packet[rule_l3_layer].dst, rule_prefix_len))
+ acl_rule = AclRule(is_permit=is_permit, proto=rule_l4_proto,
+ src_prefix=src_network,
+ dst_prefix=dst_network)
+ acl_rule.sport_from = rule_l4_sport
+ acl_rule.sport_to = rule_l4_sport
+ acl_rule.dport_from = rule_l4_dport
+ acl_rule.dport_to = rule_l4_dport
acl_rules.append(acl_rule)
if mac_type == self.WILD_MAC and ip_type == self.WILD_IP and p > 0:
continue
if is_permit:
- macip_rule = ({
- 'is_permit': is_permit,
- 'is_ipv6': is_ip6,
- 'src_ip_addr': ip_rule,
- 'src_ip_prefix_len': prefix_len,
- 'src_mac': binascii.unhexlify(mac_rule.replace(':', '')),
- 'src_mac_mask': binascii.unhexlify(
- mac_mask.replace(':', ''))})
+ macip_rule = MacipRule(
+ is_permit=is_permit,
+ src_prefix=ip_network(
+ (ip_rule, prefix_len)),
+ src_mac=MACAddress(mac_rule).packed,
+ src_mac_mask=MACAddress(mac_mask).packed)
macip_rules.append(macip_rule)
# deny all other packets
if not (mac_type == self.WILD_MAC and ip_type == self.WILD_IP):
- macip_rule = ({'is_permit': 0,
- 'is_ipv6': is_ip6,
- 'src_ip_addr': "",
- 'src_ip_prefix_len': 0,
- 'src_mac': "",
- 'src_mac_mask': ""})
+ network = IPv6Network((0, 0)) if is_ip6 else IPv4Network((0, 0))
+ macip_rule = MacipRule(
+ is_permit=0,
+ src_prefix=network,
+ src_mac=MACAddress("00:00:00:00:00:00").packed,
+ src_mac_mask=MACAddress("00:00:00:00:00:00").packed)
macip_rules.append(macip_rule)
- acl_rule = {'is_permit': 0,
- 'is_ipv6': is_ip6}
+ network = IPv6Network((0, 0)) if is_ip6 else IPv4Network((0, 0))
+ acl_rule = AclRule(is_permit=0, src_prefix=network, dst_prefix=network)
+ acl_rule.sport_from = 0
+ acl_rule.sport_to = 0
+ acl_rule.dport_from = 0
+ acl_rule.dport_to = 0
acl_rules.append(acl_rule)
return {'stream': packets,
'macip_rules': macip_rules,
if apply_rules:
if isMACIP:
- reply = self.vapi.macip_acl_add(test_dict['macip_rules'])
+ self.acl = VppMacipAcl(self, rules=test_dict['macip_rules'])
else:
- reply = self.vapi.acl_add_replace(acl_index=4294967295,
- r=test_dict['acl_rules'])
- self.assertEqual(reply.retval, 0)
- acl_index = reply.acl_index
+ self.acl = VppAcl(self, rules=test_dict['acl_rules'])
+ self.acl.add_vpp_config()
if isMACIP:
- self.vapi.macip_acl_interface_add_del(
- sw_if_index=tx_if.sw_if_index,
- acl_index=acl_index)
- reply = self.vapi.macip_acl_interface_get()
- self.assertEqual(reply.acls[tx_if.sw_if_index], acl_index)
- self.ACLS.append(reply.acls[tx_if.sw_if_index])
+ self.acl_if = VppMacipAclInterface(
+ self, sw_if_index=tx_if.sw_if_index, acls=[self.acl])
+ self.acl_if.add_vpp_config()
+
+ dump = self.acl_if.dump()
+ self.assertTrue(dump)
+ self.assertEqual(dump[0].acls[0], self.acl.acl_index)
else:
- self.vapi.acl_interface_add_del(
- sw_if_index=tx_if.sw_if_index, acl_index=acl_index)
+ self.acl_if = VppAclInterface(
+ self, sw_if_index=tx_if.sw_if_index, n_input=1,
+ acls=[self.acl])
+ self.acl_if.add_vpp_config()
else:
- self.vapi.macip_acl_interface_add_del(
- sw_if_index=tx_if.sw_if_index,
- acl_index=0)
- if try_replace:
+ if hasattr(self, "acl_if"):
+ self.acl_if.remove_vpp_config()
+ if try_replace and hasattr(self, "acl"):
if isMACIP:
- reply = self.vapi.macip_acl_add_replace(
- test_dict['macip_rules'],
- acl_index)
+ self.acl.rules = test_dict['macip_rules']
+ self.acl.add_vpp_config()
else:
- reply = self.vapi.acl_add_replace(acl_index=acl_index,
- r=test_dict['acl_rules'])
- self.assertEqual(reply.retval, 0)
+ self.acl.rules = test_dict['acl_rules']
+ self.acl.add_vpp_config()
if not isinstance(src_if, VppSubInterface):
tx_if.add_stream(test_dict['stream'])
self.get_packet_count_for_if_idx(dst_if.sw_if_index))
self.verify_capture(test_dict['stream'], capture, is_ip6)
if not isMACIP:
- self.vapi.acl_interface_add_del(sw_if_index=tx_if.sw_if_index,
- acl_index=acl_index, is_add=0)
- self.vapi.acl_del(acl_index)
+ if hasattr(self, "acl_if"):
+ self.acl_if.remove_vpp_config()
+ if hasattr(self, "acl"):
+ self.acl.remove_vpp_config()
def run_test_acls(self, mac_type, ip_type, acl_count,
rules_count, traffic=None, ip=None):
r1 = self.create_rules(acl_count=3, rules_count=[2, 2, 2])
r2 = self.create_rules(mac_type=self.OUI_MAC, ip_type=self.SUBNET_IP)
- self.apply_macip_rules(r1)
+ macip_acls = self.apply_macip_rules(r1)
acls_before = self.macip_acl_dump_debug()
# replace acls #2, #3 with new
- reply = self.vapi.macip_acl_add_replace(r2[0], 2)
- self.assertEqual(reply.retval, 0)
- self.assertEqual(reply.acl_index, 2)
- reply = self.vapi.macip_acl_add_replace(r2[1], 3)
- self.assertEqual(reply.retval, 0)
- self.assertEqual(reply.acl_index, 3)
+ macip_acls[2].rules = r2[0]
+ macip_acls[2].add_vpp_config()
+ macip_acls[3].rules = r2[1]
+ macip_acls[3].add_vpp_config()
acls_after = self.macip_acl_dump_debug()
intf_count = len(self.interfaces)+1
intf = []
- self.apply_macip_rules(self.create_rules(acl_count=3,
- rules_count=[3, 5, 4]))
+ macip_alcs = self.apply_macip_rules(
+ self.create_rules(acl_count=3, rules_count=[3, 5, 4]))
intf.append(VppLoInterface(self))
intf.append(VppLoInterface(self))
sw_if_index0 = intf[0].sw_if_index
- self.vapi.macip_acl_interface_add_del(sw_if_index0, 1)
+ macip_acl_if0 = VppMacipAclInterface(
+ self, sw_if_index=sw_if_index0, acls=[macip_alcs[1]])
+ macip_acl_if0.add_vpp_config()
reply = self.vapi.macip_acl_interface_get()
self.assertEqual(reply.count, intf_count+1)
self.assertEqual(reply.acls[sw_if_index0], 1)
sw_if_index1 = intf[1].sw_if_index
- self.vapi.macip_acl_interface_add_del(sw_if_index1, 0)
+ macip_acl_if1 = VppMacipAclInterface(
+ self, sw_if_index=sw_if_index1, acls=[macip_alcs[0]])
+ macip_acl_if1.add_vpp_config()
reply = self.vapi.macip_acl_interface_get()
self.assertEqual(reply.count, intf_count+2)
intf.append(VppLoInterface(self))
sw_if_index2 = intf[2].sw_if_index
sw_if_index3 = intf[3].sw_if_index
- self.vapi.macip_acl_interface_add_del(sw_if_index2, 1)
- self.vapi.macip_acl_interface_add_del(sw_if_index3, 1)
+ macip_acl_if2 = VppMacipAclInterface(
+ self, sw_if_index=sw_if_index2, acls=[macip_alcs[1]])
+ macip_acl_if2.add_vpp_config()
+ macip_acl_if3 = VppMacipAclInterface(
+ self, sw_if_index=sw_if_index3, acls=[macip_alcs[1]])
+ macip_acl_if3.add_vpp_config()
reply = self.vapi.macip_acl_interface_get()
self.assertEqual(reply.count, intf_count+3)
self.acl_active_table = key
self.run_verify_test(self.IP, self.IPV4, -1)
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
--- /dev/null
+from ipaddress import IPv4Network
+
+from vpp_object import VppObject
+from vpp_papi import VppEnum
+from vpp_ip import INVALID_INDEX
+from vpp_papi_provider import UnexpectedApiReturnValueError
+
+
+class VppAclPlugin(VppObject):
+
+ def __init__(self, test, enable_intf_counters=False):
+ self._test = test
+ self.enable_intf_counters = enable_intf_counters
+
+ @property
+ def enable_intf_counters(self):
+ return self._enable_intf_counters
+
+ @enable_intf_counters.setter
+ def enable_intf_counters(self, enable):
+ self.vapi.acl_stats_intf_counters_enable(enable=enable)
+
+ def add_vpp_config(self):
+ pass
+
+ def remove_vpp_config(self):
+ pass
+
+ def query_vpp_config(self):
+ pass
+
+ def object_id(self):
+ return ("acl-plugin-%d" % (self._sw_if_index))
+
+
+class AclRule():
+ """ ACL Rule """
+
+ # port ranges
+ PORTS_ALL = -1
+ PORTS_RANGE = 0
+ PORTS_RANGE_2 = 1
+ udp_sport_from = 10
+ udp_sport_to = udp_sport_from + 5
+ udp_dport_from = 20000
+ udp_dport_to = udp_dport_from + 5000
+ tcp_sport_from = 30
+ tcp_sport_to = tcp_sport_from + 5
+ tcp_dport_from = 40000
+ tcp_dport_to = tcp_dport_from + 5000
+
+ udp_sport_from_2 = 90
+ udp_sport_to_2 = udp_sport_from_2 + 5
+ udp_dport_from_2 = 30000
+ udp_dport_to_2 = udp_dport_from_2 + 5000
+ tcp_sport_from_2 = 130
+ tcp_sport_to_2 = tcp_sport_from_2 + 5
+ tcp_dport_from_2 = 20000
+ tcp_dport_to_2 = tcp_dport_from_2 + 5000
+
+ icmp4_type = 8 # echo request
+ icmp4_code = 3
+ icmp6_type = 128 # echo request
+ icmp6_code = 3
+
+ icmp4_type_2 = 8
+ icmp4_code_from_2 = 5
+ icmp4_code_to_2 = 20
+ icmp6_type_2 = 128
+ icmp6_code_from_2 = 8
+ icmp6_code_to_2 = 42
+
+ def __init__(self, is_permit, src_prefix=IPv4Network('0.0.0.0/0'),
+ dst_prefix=IPv4Network('0.0.0.0/0'),
+ proto=0, ports=PORTS_ALL):
+ self.is_permit = is_permit
+ self.src_prefix = src_prefix
+ self.dst_prefix = dst_prefix
+ self._proto = proto
+ self._ports = ports
+ self.sport_from = 0
+ self.sport_to = 0
+ self.dport_from = 0
+ self.dport_to = 0
+ self.update_ports()
+
+ def __copy__(self):
+ """
+ ports are assigned implicitly based on _proto and _ports values,
+ so we need to set them manually in case they were user defined
+ """
+ new_rule = AclRule(self.is_permit, self.src_prefix, self.dst_prefix,
+ self._proto, self._ports)
+ new_rule.sport_from = self.sport_from
+ new_rule.sport_to = self.sport_to
+ new_rule.dport_from = self.dport_from
+ new_rule.dport_to = self.dport_to
+ return new_rule
+
+ def update_ports(self):
+ if self._ports == self.PORTS_ALL:
+ self.sport_from = 0
+ self.dport_from = 0
+ self.sport_to = 65535
+ if self._proto == 1 or self._proto == 58:
+ self.sport_to = 255
+ self.dport_to = self.sport_to
+ elif self._ports == self.PORTS_RANGE:
+ if self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_ICMP:
+ self.sport_from = self.icmp4_type
+ self.sport_to = self.icmp4_type
+ self.dport_from = self.icmp4_code
+ self.dport_to = self.icmp4_code
+ elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_ICMP6:
+ self.sport_from = self.icmp6_type
+ self.sport_to = self.icmp6_type
+ self.dport_from = self.icmp6_code
+ self.dport_to = self.icmp6_code
+ elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_TCP:
+ self.sport_from = self.tcp_sport_from
+ self.sport_to = self.tcp_sport_to
+ self.dport_from = self.tcp_dport_from
+ self.dport_to = self.tcp_dport_to
+ elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP:
+ self.sport_from = self.udp_sport_from
+ self.sport_to = self.udp_sport_to
+ self.dport_from = self.udp_dport_from
+ self.dport_to = self.udp_dport_to
+ elif self._ports == self.PORTS_RANGE_2:
+ if self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_ICMP:
+ self.sport_from = self.icmp4_type_2
+ self.sport_to = self.icmp4_type_2
+ self.dport_from = self.icmp4_code_from_2
+ self.dport_to = self.icmp4_code_to_2
+ elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_ICMP6:
+ self.sport_from = self.icmp6_type_2
+ self.sport_to = self.icmp6_type_2
+ self.dport_from = self.icmp6_code_from_2
+ self.dport_to = self.icmp6_code_to_2
+ elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_TCP:
+ self.sport_from = self.tcp_sport_from_2
+ self.sport_to = self.tcp_sport_to_2
+ self.dport_from = self.tcp_dport_from_2
+ self.dport_to = self.tcp_dport_to_2
+ elif self._proto == VppEnum.vl_api_ip_proto_t.IP_API_PROTO_UDP:
+ self.sport_from = self.udp_sport_from_2
+ self.sport_to = self.udp_sport_to_2
+ self.dport_from = self.udp_dport_from_2
+ self.dport_to = self.udp_dport_to_2
+ else:
+ self.sport_from = self._ports
+ self.sport_to = self._ports
+ self.dport_from = self._ports
+ self.dport_to = self._ports
+
+ @property
+ def proto(self):
+ return self._proto
+
+ @proto.setter
+ def proto(self, proto):
+ self._proto = proto
+ self.update_ports()
+
+ @property
+ def ports(self):
+ return self._ports
+
+ @ports.setter
+ def ports(self, ports):
+ self._ports = ports
+ self.update_ports()
+
+ def encode(self):
+ return {'is_permit': self.is_permit, 'proto': self.proto,
+ 'srcport_or_icmptype_first': self.sport_from,
+ 'srcport_or_icmptype_last': self.sport_to,
+ 'src_prefix': self.src_prefix,
+ 'dstport_or_icmpcode_first': self.dport_from,
+ 'dstport_or_icmpcode_last': self.dport_to,
+ 'dst_prefix': self.dst_prefix}
+
+
+class VppAcl(VppObject):
+ """ VPP ACL """
+
+ def __init__(self, test, rules, acl_index=INVALID_INDEX, tag=None):
+ self._test = test
+ self._acl_index = acl_index
+ self.tag = tag
+ self.rules = rules
+
+ @property
+ def acl_index(self):
+ return self._acl_index
+
+ @property
+ def count(self):
+ return len(self.rules)
+
+ def encode_rules(self):
+ rules = []
+ for rule in self.rules:
+ rules.append(rule.encode())
+ return rules
+
+ def add_vpp_config(self, expect_error=False):
+ try:
+ reply = self._test.vapi.acl_add_replace(
+ acl_index=self._acl_index, tag=self.tag, count=self.count,
+ r=self.encode_rules())
+ self._acl_index = reply.acl_index
+ self._test.registry.register(self, self._test.logger)
+ if expect_error:
+ self._test.fail("Unexpected api reply")
+ return self
+ except UnexpectedApiReturnValueError:
+ if not expect_error:
+ self._test.fail("Unexpected api reply")
+ return None
+
+ def remove_vpp_config(self, expect_error=False):
+ try:
+ self._test.vapi.acl_del(acl_index=self._acl_index)
+ if expect_error:
+ self._test.fail("Unexpected api reply")
+ except UnexpectedApiReturnValueError:
+ if not expect_error:
+ self._test.fail("Unexpected api reply")
+
+ def dump(self):
+ return self._test.vapi.acl_dump(acl_index=self._acl_index)
+
+ def query_vpp_config(self):
+ dump = self.dump()
+ for rule in dump:
+ if rule.acl_index == self._acl_index:
+ return True
+ return False
+
+ def object_id(self):
+ return ("acl-%s-%d" % (self.tag, self._acl_index))
+
+
+class VppEtypeWhitelist(VppObject):
+ """ VPP Etype Whitelist """
+
+ def __init__(self, test, sw_if_index, whitelist, n_input=0):
+ self._test = test
+ self.whitelist = whitelist
+ self.n_input = n_input
+ self._sw_if_index = sw_if_index
+
+ @property
+ def sw_if_index(self):
+ return self._sw_if_index
+
+ @property
+ def count(self):
+ return len(self.whitelist)
+
+ def add_vpp_config(self):
+ self._test.vapi.acl_interface_set_etype_whitelist(
+ sw_if_index=self._sw_if_index, count=self.count,
+ n_input=self.n_input, whitelist=self.whitelist)
+ self._test.registry.register(self, self._test.logger)
+ return self
+
+ def remove_vpp_config(self):
+ self._test.vapi.acl_interface_set_etype_whitelist(
+ sw_if_index=self._sw_if_index, count=0, n_input=0, whitelist=[])
+
+ def query_vpp_config(self):
+ self._test.vapi.acl_interface_etype_whitelist_dump(
+ sw_if_index=self._sw_if_index)
+ return False
+
+ def object_id(self):
+ return ("acl-etype_wl-%d" % (self._sw_if_index))
+
+
+class VppAclInterface(VppObject):
+ """ VPP ACL Interface """
+
+ def __init__(self, test, sw_if_index, acls, n_input=0):
+ self._test = test
+ self._sw_if_index = sw_if_index
+ self.n_input = n_input
+ self.acls = acls
+
+ @property
+ def sw_if_index(self):
+ return self._sw_if_index
+
+ @property
+ def count(self):
+ return len(self.acls)
+
+ def encode_acls(self):
+ acls = []
+ for acl in self.acls:
+ acls.append(acl.acl_index)
+ return acls
+
+ def add_vpp_config(self, expect_error=False):
+ try:
+ reply = self._test.vapi.acl_interface_set_acl_list(
+ sw_if_index=self._sw_if_index, n_input=self.n_input,
+ count=self.count, acls=self.encode_acls())
+ self._test.registry.register(self, self._test.logger)
+ if expect_error:
+ self._test.fail("Unexpected api reply")
+ return self
+ except UnexpectedApiReturnValueError:
+ if not expect_error:
+ self._test.fail("Unexpected api reply")
+ return None
+
+ def remove_vpp_config(self, expect_error=False):
+ try:
+ reply = self._test.vapi.acl_interface_set_acl_list(
+ sw_if_index=self._sw_if_index, n_input=0, count=0, acls=[])
+ if expect_error:
+ self._test.fail("Unexpected api reply")
+ except UnexpectedApiReturnValueError:
+ if not expect_error:
+ self._test.fail("Unexpected api reply")
+
+ def query_vpp_config(self):
+ dump = self._test.vapi.acl_interface_list_dump(
+ sw_if_index=self._sw_if_index)
+ for acl_list in dump:
+ if acl_list.count > 0:
+ return True
+ return False
+
+ def object_id(self):
+ return ("acl-if-list-%d" % (self._sw_if_index))
+
+
+class MacipRule():
+ """ Mac Ip rule """
+
+ def __init__(self, is_permit, src_mac=0, src_mac_mask=0,
+ src_prefix=IPv4Network('0.0.0.0/0')):
+ self.is_permit = is_permit
+ self.src_mac = src_mac
+ self.src_mac_mask = src_mac_mask
+ self.src_prefix = src_prefix
+
+ def encode(self):
+ return {'is_permit': self.is_permit, 'src_mac': self.src_mac,
+ 'src_mac_mask': self.src_mac_mask,
+ 'src_prefix': self.src_prefix}
+
+
+class VppMacipAcl(VppObject):
+ """ Vpp Mac Ip ACL """
+
+ def __init__(self, test, rules, acl_index=INVALID_INDEX, tag=None):
+ self._test = test
+ self._acl_index = acl_index
+ self.tag = tag
+ self.rules = rules
+
+ @property
+ def acl_index(self):
+ return self._acl_index
+
+ @property
+ def count(self):
+ return len(self.rules)
+
+ def encode_rules(self):
+ rules = []
+ for rule in self.rules:
+ rules.append(rule.encode())
+ return rules
+
+ def add_vpp_config(self, expect_error=False):
+ try:
+ reply = self._test.vapi.macip_acl_add_replace(
+ acl_index=self._acl_index, tag=self.tag, count=self.count,
+ r=self.encode_rules())
+ self._acl_index = reply.acl_index
+ self._test.registry.register(self, self._test.logger)
+ if expect_error:
+ self._test.fail("Unexpected api reply")
+ return self
+ except UnexpectedApiReturnValueError:
+ if not expect_error:
+ self._test.fail("Unexpected api reply")
+ return None
+
+ def remove_vpp_config(self, expect_error=False):
+ try:
+ self._test.vapi.macip_acl_del(acl_index=self._acl_index)
+ if expect_error:
+ self._test.fail("Unexpected api reply")
+ except UnexpectedApiReturnValueError:
+ if not expect_error:
+ self._test.fail("Unexpected api reply")
+
+ def dump(self):
+ return self._test.vapi.macip_acl_dump(acl_index=self._acl_index)
+
+ def query_vpp_config(self):
+ dump = self.dump()
+ for rule in dump:
+ if rule.acl_index == self._acl_index:
+ return True
+ return False
+
+ def object_id(self):
+ return ("macip-acl-%s-%d" % (self.tag, self._acl_index))
+
+
+class VppMacipAclInterface(VppObject):
+ """ VPP Mac Ip ACL Interface """
+
+ def __init__(self, test, sw_if_index, acls):
+ self._test = test
+ self._sw_if_index = sw_if_index
+ self.acls = acls
+
+ @property
+ def sw_if_index(self):
+ return self._sw_if_index
+
+ @property
+ def count(self):
+ return len(self.acls)
+
+ def add_vpp_config(self):
+ for acl in self.acls:
+ self._test.vapi.macip_acl_interface_add_del(
+ is_add=True, sw_if_index=self._sw_if_index,
+ acl_index=acl.acl_index)
+ self._test.registry.register(self, self._test.logger)
+
+ def remove_vpp_config(self):
+ for acl in self.acls:
+ self._test.vapi.macip_acl_interface_add_del(
+ is_add=False, sw_if_index=self._sw_if_index,
+ acl_index=acl.acl_index)
+
+ def dump(self):
+ return self._test.vapi.macip_acl_interface_list_dump(
+ sw_if_index=self._sw_if_index)
+
+ def query_vpp_config(self):
+ dump = self.dump()
+ for acl_list in dump:
+ for acl_index in acl_list.acls:
+ if acl_index != INVALID_INDEX:
+ return True
+ return False
+
+ def object_id(self):
+ return ("macip-acl-if-list-%d" % (self._sw_if_index))
from vpp_vxlan_gbp_tunnel import find_vxlan_gbp_tunnel, INDEX_INVALID, \
VppVxlanGbpTunnel
from vpp_neighbor import VppNeighbor
+from vpp_acl import AclRule, VppAcl
try:
text_type = unicode
except NameError:
return find_gbp_vxlan(self._test, self.vni)
-class VppGbpAcl(VppObject):
- """
- GBP Acl
- """
-
- def __init__(self, test):
- self._test = test
- self.acl_index = 4294967295
-
- def create_rule(self, is_ipv6=0, permit_deny=0, proto=-1,
- s_prefix=0, s_ip=b'\x00\x00\x00\x00', sport_from=0,
- sport_to=65535, d_prefix=0, d_ip=b'\x00\x00\x00\x00',
- dport_from=0, dport_to=65535):
- if proto == -1 or proto == 0:
- sport_to = 0
- dport_to = sport_to
- elif proto == 1 or proto == 58:
- sport_to = 255
- dport_to = sport_to
- rule = ({'is_permit': permit_deny, 'is_ipv6': is_ipv6, 'proto': proto,
- 'srcport_or_icmptype_first': sport_from,
- 'srcport_or_icmptype_last': sport_to,
- 'src_ip_prefix_len': s_prefix,
- 'src_ip_addr': s_ip,
- 'dstport_or_icmpcode_first': dport_from,
- 'dstport_or_icmpcode_last': dport_to,
- 'dst_ip_prefix_len': d_prefix,
- 'dst_ip_addr': d_ip})
- return rule
-
- def add_vpp_config(self, rules):
-
- reply = self._test.vapi.acl_add_replace(acl_index=self.acl_index,
- r=rules,
- tag=b'GBPTest')
- self.acl_index = reply.acl_index
- return self.acl_index
-
- def remove_vpp_config(self):
- self._test.vapi.acl_del(self.acl_index)
-
- def object_id(self):
- return "gbp-acl:[%d]" % (self.acl_index)
-
- def query_vpp_config(self):
- cs = self._test.vapi.acl_dump()
- for c in cs:
- if c.acl_index == self.acl_index:
- return True
- return False
-
-
class TestGBP(VppTestCase):
""" GBP Test Case """
#
# A uni-directional contract from EPG 220 -> 221
#
- acl = VppGbpAcl(self)
- rule = acl.create_rule(permit_deny=1, proto=17)
- rule2 = acl.create_rule(is_ipv6=1, permit_deny=1, proto=17)
- acl_index = acl.add_vpp_config([rule, rule2])
+ rule = AclRule(is_permit=1, proto=17)
+ rule2 = AclRule(src_prefix=IPv6Network((0, 0)),
+ dst_prefix=IPv6Network((0, 0)), is_permit=1, proto=17)
+ acl = VppAcl(self, rules=[rule, rule2])
+ acl.add_vpp_config()
+
c1 = VppGbpContract(
- self, 400, epgs[0].sclass, epgs[1].sclass, acl_index,
+ self, 400, epgs[0].sclass, epgs[1].sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# contract for the return direction
#
c2 = VppGbpContract(
- self, 400, epgs[1].sclass, epgs[0].sclass, acl_index,
+ self, 400, epgs[1].sclass, epgs[0].sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# A uni-directional contract from EPG 220 -> 222 'L3 routed'
#
c3 = VppGbpContract(
- self, 400, epgs[0].sclass, epgs[2].sclass, acl_index,
+ self, 400, epgs[0].sclass, epgs[2].sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# no policy yet
self.send_and_assert_no_replies(eps[0].itf,
pkt_inter_epg_220_to_global * NUM_PKTS)
+ rule = AclRule(is_permit=1, proto=17, ports=1234)
+ rule2 = AclRule(is_permit=1, proto=17, ports=1234,
+ src_prefix=IPv6Network((0, 0)),
+ dst_prefix=IPv6Network((0, 0)))
+ acl2 = VppAcl(self, rules=[rule, rule2])
+ acl2.add_vpp_config()
- acl2 = VppGbpAcl(self)
- rule = acl2.create_rule(permit_deny=1, proto=17, sport_from=1234,
- sport_to=1234, dport_from=1234, dport_to=1234)
- rule2 = acl2.create_rule(is_ipv6=1, permit_deny=1, proto=17,
- sport_from=1234, sport_to=1234,
- dport_from=1234, dport_to=1234)
-
- acl_index2 = acl2.add_vpp_config([rule, rule2])
c4 = VppGbpContract(
- self, 400, epgs[0].sclass, epgs[3].sclass, acl_index2,
+ self, 400, epgs[0].sclass, epgs[3].sclass, acl2.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
self.pg7, pkt_inter_epg_220_from_global * NUM_PKTS)
c5 = VppGbpContract(
- self, 400, epgs[3].sclass, epgs[0].sclass, acl_index2,
+ self, 400, epgs[3].sclass, epgs[0].sclass, acl2.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
#
# Add the contract so they can talk
#
- acl = VppGbpAcl(self)
- rule = acl.create_rule(permit_deny=1, proto=17)
- rule2 = acl.create_rule(is_ipv6=1, permit_deny=1, proto=17)
- acl_index = acl.add_vpp_config([rule, rule2])
+ rule = AclRule(is_permit=1, proto=17)
+ rule2 = AclRule(src_prefix=IPv6Network((0, 0)),
+ dst_prefix=IPv6Network((0, 0)), is_permit=1, proto=17)
+ acl = VppAcl(self, rules=[rule, rule2])
+ acl.add_vpp_config()
+
c1 = VppGbpContract(
- self, 401, epg_220.sclass, epg_330.sclass, acl_index,
+ self, 401, epg_220.sclass, epg_330.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
self.assertFalse(rx[VXLAN].gpflags.A)
self.assertFalse(rx[VXLAN].gpflags.D)
- acl = VppGbpAcl(self)
- rule = acl.create_rule(permit_deny=1, proto=17)
- rule2 = acl.create_rule(is_ipv6=1, permit_deny=1, proto=17)
- acl_index = acl.add_vpp_config([rule, rule2])
+ rule = AclRule(is_permit=1, proto=17)
+ rule2 = AclRule(src_prefix=IPv6Network((0, 0)),
+ dst_prefix=IPv6Network((0, 0)), is_permit=1, proto=17)
+ acl = VppAcl(self, rules=[rule, rule2])
+ acl.add_vpp_config()
+
c2 = VppGbpContract(
- self, 401, epg_330.sclass, epg_220.sclass, acl_index,
+ self, 401, epg_330.sclass, epg_220.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
#
# A uni-directional contract from EPG 220 -> 221
#
- acl = VppGbpAcl(self)
- rule = acl.create_rule(permit_deny=1, proto=17)
- rule2 = acl.create_rule(is_ipv6=1, permit_deny=1, proto=17)
- rule3 = acl.create_rule(permit_deny=1, proto=1)
- acl_index = acl.add_vpp_config([rule, rule2, rule3])
+ rule = AclRule(is_permit=1, proto=17)
+ rule2 = AclRule(src_prefix=IPv6Network((0, 0)),
+ dst_prefix=IPv6Network((0, 0)), is_permit=1, proto=17)
+ rule3 = AclRule(is_permit=1, proto=1)
+ acl = VppAcl(self, rules=[rule, rule2, rule3])
+ acl.add_vpp_config()
+
c1 = VppGbpContract(
- self, 400, epgs[0].sclass, epgs[1].sclass, acl_index,
+ self, 400, epgs[0].sclass, epgs[1].sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# contract for the return direction
#
c2 = VppGbpContract(
- self, 400, epgs[1].sclass, epgs[0].sclass, acl_index,
+ self, 400, epgs[1].sclass, epgs[0].sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# contract between 220 and 222 uni-direction
#
c3 = VppGbpContract(
- self, 400, epgs[0].sclass, epgs[2].sclass, acl_index,
+ self, 400, epgs[0].sclass, epgs[2].sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# Add a contract with a rule to load-balance redirect via SEP1 and SEP2
# one of the next-hops is via an EP that is not known
#
- acl = VppGbpAcl(self)
- rule4 = acl.create_rule(permit_deny=1, proto=17)
- rule6 = acl.create_rule(is_ipv6=1, permit_deny=1, proto=17)
- acl_index = acl.add_vpp_config([rule4, rule6])
+ rule4 = AclRule(is_permit=1, proto=17)
+ rule6 = AclRule(src_prefix=IPv6Network((0, 0)),
+ dst_prefix=IPv6Network((0, 0)), is_permit=1, proto=17)
+ acl = VppAcl(self, rules=[rule4, rule6])
+ acl.add_vpp_config()
#
# test the src-ip hash mode
#
c1 = VppGbpContract(
- self, 402, epg_220.sclass, epg_222.sclass, acl_index,
+ self, 402, epg_220.sclass, epg_222.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
c1.add_vpp_config()
c2 = VppGbpContract(
- self, 402, epg_222.sclass, epg_220.sclass, acl_index,
+ self, 402, epg_222.sclass, epg_220.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# test the symmetric hash mode
#
c1 = VppGbpContract(
- self, 402, epg_220.sclass, epg_222.sclass, acl_index,
+ self, 402, epg_220.sclass, epg_222.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SYMMETRIC,
c1.add_vpp_config()
c2 = VppGbpContract(
- self, 402, epg_222.sclass, epg_220.sclass, acl_index,
+ self, 402, epg_222.sclass, epg_220.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SYMMETRIC,
Raw(b'\xa5' * 100))]
c3 = VppGbpContract(
- self, 402, epg_220.sclass, epg_221.sclass, acl_index,
+ self, 402, epg_220.sclass, epg_221.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SYMMETRIC,
vx_tun_l3.add_vpp_config()
c4 = VppGbpContract(
- self, 402, epg_221.sclass, epg_220.sclass, acl_index,
+ self, 402, epg_221.sclass, epg_220.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# test the dst-ip hash mode
#
c5 = VppGbpContract(
- self, 402, epg_220.sclass, epg_221.sclass, acl_index,
+ self, 402, epg_220.sclass, epg_221.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_DST_IP,
# contract redirecting to sep5
VppGbpContract(
- self, 402, 4220, 4221, acl_index,
+ self, 402, 4220, 4221, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_DST_IP,
# change the contract between l3out to redirect to local SEPs
# instead of remote SEP
VppGbpContract(
- self, 402, 4220, 4221, acl_index,
+ self, 402, 4220, 4221, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_DST_IP,
# contract to redirect to learnt SEP
VppGbpContract(
- self, 402, epg_221.sclass, epg_222.sclass, acl_index,
+ self, 402, epg_221.sclass, epg_222.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_DST_IP,
# Add a contract with a rule to load-balance redirect via SEP1 and SEP2
# one of the next-hops is via an EP that is not known
#
- acl = VppGbpAcl(self)
- rule4 = acl.create_rule(permit_deny=1, proto=17)
- rule6 = acl.create_rule(is_ipv6=1, permit_deny=1, proto=17)
- acl_index = acl.add_vpp_config([rule4, rule6])
+ rule4 = AclRule(is_permit=1, proto=17)
+ rule6 = AclRule(src_prefix=IPv6Network((0, 0)),
+ dst_prefix=IPv6Network((0, 0)), is_permit=1, proto=17)
+ acl = VppAcl(self, rules=[rule4, rule6])
+ acl.add_vpp_config()
#
# test the src-ip hash mode
#
c1 = VppGbpContract(
- self, 402, epg_220.sclass, epg_221.sclass, acl_index,
+ self, 402, epg_220.sclass, epg_221.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SYMMETRIC,
c1.add_vpp_config()
c2 = VppGbpContract(
- self, 402, epg_221.sclass, epg_220.sclass, acl_index,
+ self, 402, epg_221.sclass, epg_220.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_REDIRECT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SYMMETRIC,
# contract for SEP to communicate with dst EP
c3 = VppGbpContract(
- self, 402, epg_320.sclass, epg_221.sclass, acl_index,
+ self, 402, epg_320.sclass, epg_221.sclass, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SYMMETRIC),
#
# contract for the external nets to communicate
#
- acl = VppGbpAcl(self)
- rule4 = acl.create_rule(permit_deny=1, proto=17)
- rule6 = acl.create_rule(is_ipv6=1, permit_deny=1, proto=17)
- acl_index = acl.add_vpp_config([rule4, rule6])
+ rule4 = AclRule(is_permit=1, proto=17)
+ rule6 = AclRule(src_prefix=IPv6Network((0, 0)),
+ dst_prefix=IPv6Network((0, 0)), is_permit=1, proto=17)
+ acl = VppAcl(self, rules=[rule4, rule6])
+ acl.add_vpp_config()
#
# A contract with the wrong scope is not matched
#
c_44 = VppGbpContract(
- self, 44, 4220, 4221, acl_index,
+ self, 44, 4220, 4221, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
self.send_and_assert_no_replies(self.pg0, p * 1)
c1 = VppGbpContract(
- self, 55, 4220, 4221, acl_index,
+ self, 55, 4220, 4221, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# Contracts allowing ext-net 200 to talk with external EPs
#
c2 = VppGbpContract(
- self, 55, 4220, 113, acl_index,
+ self, 55, 4220, 113, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
[ETH_P_IP, ETH_P_IPV6])
c2.add_vpp_config()
c3 = VppGbpContract(
- self, 55, 113, 4220, acl_index,
+ self, 55, 113, 4220, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# Add contracts ext-nets for 220 -> 222
#
c4 = VppGbpContract(
- self, 55, 4220, 4222, acl_index,
+ self, 55, 4220, 4222, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
#
# contract for the external nets to communicate
#
- acl = VppGbpAcl(self)
- rule4 = acl.create_rule(permit_deny=1, proto=17)
- rule6 = acl.create_rule(is_ipv6=1, permit_deny=1, proto=17)
- acl_index = acl.add_vpp_config([rule4, rule6])
+ rule4 = AclRule(is_permit=1, proto=17)
+ rule6 = AclRule(src_prefix=IPv6Network((0, 0)),
+ dst_prefix=IPv6Network((0, 0)), is_permit=1, proto=17)
+ acl = VppAcl(self, rules=[rule4, rule6])
+ acl.add_vpp_config()
c1 = VppGbpContract(
- self, 55, 4220, 4221, acl_index,
+ self, 55, 4220, 4221, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# Contracts allowing ext-net 200 to talk with external EPs
#
c2 = VppGbpContract(
- self, 55, 4220, 113, acl_index,
+ self, 55, 4220, 113, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
[ETH_P_IP, ETH_P_IPV6])
c2.add_vpp_config()
c3 = VppGbpContract(
- self, 55, 113, 4220, acl_index,
+ self, 55, 113, 4220, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
# Add contracts ext-nets for 220 -> 222
#
c4 = VppGbpContract(
- self, 55, 4220, 4222, acl_index,
+ self, 55, 4220, 4222, acl.acl_index,
[VppGbpContractRule(
VppEnum.vl_api_gbp_rule_action_t.GBP_API_RULE_PERMIT,
VppEnum.vl_api_gbp_hash_mode_t.GBP_API_HASH_MODE_SRC_IP,
from ipaddress import IPv6Network
from util import ppc, ppp
from socket import inet_pton, AF_INET
+from vpp_acl import AclRule, VppAcl, VppAclInterface
# NAT HA protocol event data
self.verify_capture_in(capture, self.pg0)
# Create an ACL blocking everything
- out2in_deny_rule = {
- 'is_permit': 0,
- 'is_ipv6': 0,
- 'src_ip_addr': inet_pton(AF_INET, "0.0.0.0"),
- 'src_ip_prefix_len': 0,
- 'dst_ip_addr': inet_pton(AF_INET, "0.0.0.0"),
- '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,
- }
- out2in_rules = [out2in_deny_rule]
- res = self.vapi.acl_add_replace(0xffffffff, out2in_rules)
- self.assertEqual(res.retval, 0, "error adding out2in ACL")
- out2in_acl = res.acl_index
+ out2in_deny_rule = AclRule(is_permit=0)
+ out2in_acl = VppAcl(self, rules=[out2in_deny_rule])
+ out2in_acl.add_vpp_config()
+
+ # create an ACL to permit/reflect everything
+ in2out_reflect_rule = AclRule(is_permit=2)
+ in2out_acl = VppAcl(self, rules=[in2out_reflect_rule])
+ in2out_acl.add_vpp_config()
# apply as input acl on interface and confirm it blocks everything
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
- n_input=1,
- acls=[out2in_acl])
+ acl_if = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index,
+ n_input=1, acls=[out2in_acl])
+ acl_if.add_vpp_config()
self.send_and_assert_no_replies(self.pg1, pkts_out2in)
- # create an ACL to permit/reflect everything
- in2out_reflect_rule = {
- 'is_permit': 2,
- 'is_ipv6': 0,
- 'src_ip_addr': inet_pton(AF_INET, "0.0.0.0"),
- 'src_ip_prefix_len': 0,
- 'dst_ip_addr': inet_pton(AF_INET, "0.0.0.0"),
- '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,
- }
- in2out_rules = [in2out_reflect_rule]
- res = self.vapi.acl_add_replace(0xffffffff, in2out_rules)
- self.assertEqual(res.retval, 0, "error adding in2out ACL")
- in2out_acl = res.acl_index
-
# apply output acl
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
- n_input=1,
- acls=[out2in_acl, in2out_acl])
+ acl_if.acls = [out2in_acl, in2out_acl]
+ acl_if.add_vpp_config()
# send in2out to generate ACL state (NAT state was created earlier)
capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1,
len(pkts_in2out))
self.verify_capture_in(capture, self.pg0)
self.logger.info(self.vapi.cli("show trace"))
- # Clean up
- # Remove ACLs from interface
- self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
- n_input=0,
- acls=[])
- # delete ACLs
- self.vapi.acl_del(acl_index=out2in_acl, expected_retval=0)
- self.vapi.acl_del(acl_index=in2out_acl, expected_retval=0)
-
def test_multiple_vrf(self):
""" Multiple VRF setup """
external_addr = '1.2.3.4'
from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathType
from vpp_l2 import L2_PORT_TYPE
from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint
+from vpp_acl import AclRule, VppAcl, VppAclInterface
from scapy.packet import Raw
from scapy.layers.l2 import Ether, Dot1Q
from scapy.layers.inet import IP, UDP
from socket import AF_INET, inet_pton
+from ipaddress import IPv4Network
NUM_PKTS = 67
#
# Add an output L3 ACL that will block the traffic
#
- rule_1 = ({'is_permit': 0,
- 'is_ipv6': 0,
- 'proto': 17,
- 'srcport_or_icmptype_first': 1234,
- 'srcport_or_icmptype_last': 1234,
- 'src_ip_prefix_len': 32,
- 'src_ip_addr': inet_pton(AF_INET, any_src_addr),
- 'dstport_or_icmpcode_first': 1234,
- 'dstport_or_icmpcode_last': 1234,
- 'dst_ip_prefix_len': 32,
- 'dst_ip_addr': inet_pton(AF_INET, ip_non_tag_bridged)})
- acl = self.vapi.acl_add_replace(acl_index=4294967295,
- r=[rule_1])
+ rule_1 = AclRule(is_permit=0, proto=17, ports=1234,
+ src_prefix=IPv4Network((any_src_addr, 32)),
+ dst_prefix=IPv4Network((ip_non_tag_bridged, 32)))
+ acl = VppAcl(self, rules=[rule_1])
+ acl.add_vpp_config()
#
# Apply the ACL on the output interface
#
- self.vapi.acl_interface_set_acl_list(self.pg1.sw_if_index,
- 0,
- [acl.acl_index])
+ acl_if1 = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index,
+ n_input=0, acls=[acl])
+ acl_if1.add_vpp_config()
#
# Send packet's that should match the ACL and be dropped
#
# cleanup
#
- self.vapi.acl_interface_set_acl_list(self.pg1.sw_if_index,
- 0, [])
- self.vapi.acl_del(acl.acl_index)
+ acl_if1.remove_vpp_config()
+ acl.remove_vpp_config()
self.vapi.sw_interface_set_l2_bridge(
rx_sw_if_index=self.pg0.sw_if_index, bd_id=1, enable=0)
#!/usr/bin/env python3
from socket import AF_INET, AF_INET6, inet_pton
import unittest
+from ipaddress import IPv4Network
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from framework import VppTestCase, VppTestRunner
from vpp_interface import VppInterface
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
+from vpp_acl import AclRule, VppAcl, VppAclInterface
NUM_PKTS = 67
#
# Attach ACL to ensure features are run on the pipe
#
- rule_1 = ({'is_permit': 0,
- 'is_ipv6': 0,
- 'proto': 17,
- 'srcport_or_icmptype_first': 1234,
- 'srcport_or_icmptype_last': 1234,
- 'src_ip_prefix_len': 32,
- 'src_ip_addr': inet_pton(AF_INET, "1.1.1.1"),
- 'dstport_or_icmpcode_first': 1234,
- 'dstport_or_icmpcode_last': 1234,
- 'dst_ip_prefix_len': 32,
- 'dst_ip_addr': inet_pton(AF_INET, "1.1.1.2")})
- acl = self.vapi.acl_add_replace(acl_index=4294967295,
- r=[rule_1])
+ rule_1 = AclRule(is_permit=0, proto=17,
+ src_prefix=IPv4Network("1.1.1.1/32"),
+ dst_prefix=IPv4Network("1.1.1.2/32"), ports=1234)
+ acl = VppAcl(self, rules=[rule_1])
+ acl.add_vpp_config()
# Apply the ACL on the pipe on output
- self.vapi.acl_interface_set_acl_list(pipes[0].east,
- 0,
- [acl.acl_index])
+ acl_if_e = VppAclInterface(self, sw_if_index=pipes[0].east, n_input=0,
+ acls=[acl])
+ acl_if_e.add_vpp_config()
+
self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS)
self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)
# remove from output and apply on input
- self.vapi.acl_interface_set_acl_list(pipes[0].east,
- 0,
- [])
- self.vapi.acl_interface_set_acl_list(pipes[0].west,
- 1,
- [acl.acl_index])
+ acl_if_e.remove_vpp_config()
+ acl_if_w = VppAclInterface(self, sw_if_index=pipes[0].west, n_input=1,
+ acls=[acl])
+ acl_if_w.add_vpp_config()
+
self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS)
self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)
- self.vapi.acl_interface_set_acl_list(pipes[0].west,
- 0,
- [])
+
+ acl_if_w.remove_vpp_config()
self.send_and_expect(self.pg0, p * NUM_PKTS, self.pg1)
self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)
#
# Use ACLs to test features run on the Pipes
#
- self.vapi.acl_interface_set_acl_list(pipes[1].east,
- 0,
- [acl.acl_index])
+ acl_if_e1 = VppAclInterface(self, sw_if_index=pipes[1].east, n_input=0,
+ acls=[acl])
+ acl_if_e1.add_vpp_config()
self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS)
self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)
# remove from output and apply on input
- self.vapi.acl_interface_set_acl_list(pipes[1].east,
- 0,
- [])
- self.vapi.acl_interface_set_acl_list(pipes[1].west,
- 1,
- [acl.acl_index])
+ acl_if_e1.remove_vpp_config()
+ acl_if_w1 = VppAclInterface(self, sw_if_index=pipes[1].west, n_input=1,
+ acls=[acl])
+ acl_if_w1.add_vpp_config()
self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS)
self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)
- self.vapi.acl_interface_set_acl_list(pipes[1].west,
- 0,
- [])
+ acl_if_w1.remove_vpp_config()
+
self.send_and_expect(self.pg2, p_east * NUM_PKTS, self.pg3)
self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)
return self.api(self.papi.sr_mpls_policy_del,
{'bsid': bsid})
- def acl_add_replace(self, acl_index, r, tag='',
- expected_retval=0):
- """Add/replace an ACL
- :param int acl_index: ACL index to replace, 2^32-1 to create new ACL.
- :param acl_rule r: ACL rules array.
- :param str tag: symbolic tag (description) for this ACL.
- :param int count: number of rules.
- """
- return self.api(self.papi.acl_add_replace,
- {'acl_index': acl_index,
- 'r': r,
- 'count': len(r),
- 'tag': tag},
- expected_retval=expected_retval)
-
- def acl_del(self, acl_index, expected_retval=0):
- """
-
- :param acl_index:
- :return:
- """
- return self.api(self.papi.acl_del,
- {'acl_index': acl_index},
- expected_retval=expected_retval)
-
- def acl_interface_set_acl_list(self, sw_if_index, n_input, acls,
- expected_retval=0):
- return self.api(self.papi.acl_interface_set_acl_list,
- {'sw_if_index': sw_if_index,
- 'count': len(acls),
- 'n_input': n_input,
- 'acls': acls},
- expected_retval=expected_retval)
-
- def acl_interface_set_etype_whitelist(self, sw_if_index,
- n_input, whitelist,
- expected_retval=0):
- return self.api(self.papi.acl_interface_set_etype_whitelist,
- {'sw_if_index': sw_if_index,
- 'count': len(whitelist),
- 'n_input': n_input,
- 'whitelist': whitelist},
- expected_retval=expected_retval)
-
- def acl_interface_add_del(self,
- sw_if_index,
- acl_index,
- is_add=1):
- """ Add/Delete ACL to/from interface
-
- :param sw_if_index:
- :param acl_index:
- :param is_add: (Default value = 1)
- """
-
- return self.api(self.papi.acl_interface_add_del,
- {'is_add': is_add,
- 'is_input': 1,
- 'sw_if_index': sw_if_index,
- 'acl_index': acl_index})
-
- def acl_dump(self, acl_index, expected_retval=0):
- return self.api(self.papi.acl_dump,
- {'acl_index': acl_index},
- expected_retval=expected_retval)
-
- def acl_interface_list_dump(self, sw_if_index=0xFFFFFFFF,
- expected_retval=0):
- return self.api(self.papi.acl_interface_list_dump,
- {'sw_if_index': sw_if_index},
- expected_retval=expected_retval)
-
- def macip_acl_add(self, rules, tag=""):
- """ Add MACIP acl
-
- :param rules: list of rules for given acl
- :param tag: acl tag
- """
-
- return self.api(self.papi.macip_acl_add,
- {'r': rules,
- 'count': len(rules),
- 'tag': tag})
-
- def macip_acl_add_replace(self, rules, acl_index=0xFFFFFFFF, tag=""):
- """ Add MACIP acl
-
- :param rules: list of rules for given acl
- :param tag: acl tag
- """
-
- return self.api(self.papi.macip_acl_add_replace,
- {'acl_index': acl_index,
- 'r': rules,
- 'count': len(rules),
- 'tag': tag})
-
- def macip_acl_interface_add_del(self,
- sw_if_index,
- acl_index,
- is_add=1):
- """ Add MACIP acl to interface
-
- :param sw_if_index:
- :param acl_index:
- :param is_add: (Default value = 1)
- """
-
- return self.api(self.papi.macip_acl_interface_add_del,
- {'is_add': is_add,
- 'sw_if_index': sw_if_index,
- 'acl_index': acl_index})
-
- def macip_acl_dump(self, acl_index=4294967295):
- """ Return MACIP acl dump
- """
-
- return self.api(
- self.papi.macip_acl_dump, {'acl_index': acl_index})
-
def ip_punt_police(self,
policer_index,
is_ip6=0,
Code Review / vpp.git / commitdiff