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].src_prefix.address.af,
- (uint8_t*)&payload.r[ii].src_prefix.address.un,
- payload.r[ii].src_prefix.len);
+ 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,
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].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);
+ 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);
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 = (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);
+ payload.is_permit = rule.action().value();
+ rule.src_ip().to_vpp(
+ &payload.is_ipv6, payload.src_ip_addr, &payload.src_ip_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 = (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.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.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();
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 = 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()
+ 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])
#
# ABF policy for ACL 1 - path via interface 1
#
# 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()
+ 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])
#
# ABF policy for ACL 1 - path via interface 1
used to control the ACL plugin
*/
-option version = "2.0.0";
+option version = "1.0.1";
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 */
- string tag[64]; /* What gets in here gets out in the corresponding tag field when dumping the ACLs. */
+ u8 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;
- bool is_add [default=true];
+ u8 is_add;
/*
* is_input = 0 => ACL applied on interface egress
* is_input = 1 => ACL applied on interface ingress
*/
- bool is_input;
- vl_api_interface_index_t sw_if_index;
+ u8 is_input;
+ u32 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;
- vl_api_interface_index_t sw_if_index;
+ u32 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;
- string tag[64]; /* Same blob that was supplied to us when creating the ACL, one hopes. */
+ u8 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;
- vl_api_interface_index_t sw_if_index; /* ~0 for all interfaces */
+ u32 sw_if_index; /* ~0 for all interfaces */
option vat_help = "[<intfc> | sw_if_index <if-idx>]";
};
define acl_interface_list_details
{
u32 context;
- vl_api_interface_index_t sw_if_index;
+ u32 sw_if_index;
u8 count;
u8 n_input;
u32 acls[count];
{
u32 client_index;
u32 context;
- string tag[64];
+ u8 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 */
- string tag[64];
+ u8 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;
- bool is_add [default=true];
+ u8 is_add;
/* MACIP ACLs are always input */
- vl_api_interface_index_t sw_if_index;
+ u32 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;
- string tag[64];
+ u8 tag[64];
u32 count;
vl_api_macip_acl_rule_t r[count];
};
{
u32 client_index;
u32 context;
- vl_api_interface_index_t sw_if_index; /* ~0 for all interfaces */
+ u32 sw_if_index; /* ~0 for all interfaces */
};
/** \brief Details about a single MACIP ACL contents
define macip_acl_interface_list_details
{
u32 context;
- vl_api_interface_index_t sw_if_index;
+ u32 sw_if_index;
u8 count;
u32 acls[count];
};
{
u32 client_index;
u32 context;
- vl_api_interface_index_t sw_if_index;
+ u32 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;
- vl_api_interface_index_t sw_if_index; /* ~0 for all interfaces */
+ u32 sw_if_index; /* ~0 for all interfaces */
option vat_help = "[<intfc> | sw_if_index <if-idx>]";
};
define acl_interface_etype_whitelist_details
{
u32 context;
- vl_api_interface_index_t sw_if_index;
+ u32 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 (const vl_api_prefix_t * prefix)
+acl_api_ip4_invalid_prefix (void *ip4_pref_raw, u8 ip4_prefix_len)
{
ip4_address_t ip4_addr;
ip4_address_t ip4_mask;
ip4_address_t ip4_masked_addr;
- if (prefix->len > 32)
- return 1;
-
- ip4_address_decode (prefix->address.un.ip4, &ip4_addr);
- ip4_preflen_to_mask (prefix->len, &ip4_mask);
+ memcpy (&ip4_addr, ip4_pref_raw, sizeof (ip4_addr));
+ ip4_preflen_to_mask (ip4_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, prefix->address.un.ip4, prefix->len,
- format_ip4_address, &ip4_masked_addr);
+ format_ip4_address, ip4_pref_raw, ip4_prefix_len, format_ip4_address,
+ &ip4_masked_addr);
}
return ret;
}
static int
-acl_api_ip6_invalid_prefix (const vl_api_prefix_t * prefix)
+acl_api_ip6_invalid_prefix (void *ip6_pref_raw, u8 ip6_prefix_len)
{
ip6_address_t ip6_addr;
ip6_address_t ip6_mask;
ip6_address_t ip6_masked_addr;
- if (prefix->len > 128)
- return 1;
-
- ip6_address_decode (prefix->address.un.ip6, &ip6_addr);
- ip6_preflen_to_mask (prefix->len, &ip6_mask);
+ memcpy (&ip6_addr, ip6_pref_raw, sizeof (ip6_addr));
+ ip6_preflen_to_mask (ip6_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, prefix->address.un.ip6, prefix->len,
- format_ip6_address, &ip6_masked_addr);
+ format_ip6_address, ip6_pref_raw, 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 (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 (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 (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].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->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->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].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;
+ 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;
}
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;
- 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->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;
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;
- 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;
+ 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;
}
}
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->src_prefix.address.af ? AF_INET6 : AF_INET;
+ int af = a->is_ipv6 ? AF_INET6 : AF_INET;
u8 src[INET6_ADDRSTRLEN];
u8 dst[INET6_ADDRSTRLEN];
- inet_ntop(af, &a->src_prefix.address.un, (void *)src, sizeof(src));
- inet_ntop(af, &a->dst_prefix.address.un, (void *)dst, sizeof(dst));
+ inet_ntop(af, a->src_ip_addr, (void *)src, sizeof(src));
+ inet_ntop(af, a->dst_ip_addr, (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->src_prefix.address.af ? "ipv6" : "ipv4", a->is_permit,
- src, a->src_prefix.len,
- dst, a->dst_prefix.len,
+ a->is_ipv6 ? "ipv6" : "ipv4", a->is_permit,
+ src, a->src_ip_prefix_len,
+ dst, a->dst_ip_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->src_prefix.address.af ? AF_INET6 : AF_INET;
+ int af = a->is_ipv6 ? AF_INET6 : AF_INET;
u8 src[INET6_ADDRSTRLEN];
- inet_ntop(af, &a->src_prefix.address.un, (void *)src, sizeof(src));
+ inet_ntop(af, a->src_ip_addr, (void *)src, sizeof(src));
out = format(out, "%s action %d ip %s/%d mac %U mask %U",
- a->src_prefix.address.af ? "ipv6" : "ipv4", a->is_permit,
- src, a->src_prefix.len,
+ a->is_ipv6 ? "ipv6" : "ipv4", a->is_permit,
+ src, a->src_ip_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, "permit+reflect"))
+ 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"))
{
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_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].src_ip_addr, &src_v4address, 4);
+ rules[rule_idx].src_ip_prefix_len = src_prefix_length;
+ rules[rule_idx].is_ipv6 = 0;
}
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_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;
+ 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;
}
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_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;
+ 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;
}
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_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;
+ 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;
}
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_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;
+ 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;
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_prefix.address.un.ip4, &src_v4address, 4);
- rules[rule_idx].src_prefix.address.af = ADDRESS_IP4;
- rules[rule_idx].src_prefix.len = 0;
+ memcpy (rules[rule_idx].src_ip_addr, &src_v4address, 4);
+ rules[rule_idx].src_ip_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_prefix.address.un.ip4, &dst_v4address, 4);
- rules[rule_idx].dst_prefix.address.af = ADDRESS_IP4;
- rules[rule_idx].dst_prefix.len = 0;
+ memcpy (rules[rule_idx].dst_ip_addr, &dst_v4address, 4);
+ rules[rule_idx].dst_ip_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, "permit"))
+ 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"))
{
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_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].src_ip_addr, &src_v4address, 4);
+ rules[rule_idx].src_ip_prefix_len = src_prefix_length;
+ rules[rule_idx].is_ipv6 = 0;
}
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_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;
+ 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;
}
else if (unformat (i, "mac %U",
my_unformat_mac_address, &src_mac))
while (unformat_check_input (i) != UNFORMAT_END_OF_INPUT)
{
- if (unformat (i, "permit"))
+ 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"))
{
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_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;
+ 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;
}
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_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;
+ 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;
}
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 src_prefix - Source prefix
- @param dst_prefix - Destination prefix
+ @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 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
{
- vl_api_acl_action_t is_permit;
- vl_api_prefix_t src_prefix;
- vl_api_prefix_t dst_prefix;
+ 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;
/*
* L4 protocol. IANA number. 1 = ICMP, 58 = ICMPv6, 6 = TCP, 17 = UDP.
* 0 => ignore L4 and ignore the ports/tcpflags when matching.
*/
- vl_api_ip_proto_t proto;
+ u8 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_prefix - Source prefix value
+ @param src_ip_addr - Source prefix value
+ @param src_ip_prefix_len - Source prefix length
*/
typedef macip_acl_rule
{
- vl_api_acl_action_t is_permit;
+ u8 is_permit;
+ u8 is_ipv6;
/*
* The source mac of the packet ANDed with src_mac_mask.
* The source ip[46] address in the packet is matched
- * against src_prefix set to 0.
+ * against src_ip_addr, with src_ip_prefix_len set to 0.
*
* For better performance, minimize the number of
- * (src_mac_mask, src_prefix.len) combinations
+ * (src_mac_mask, src_ip_prefix_len) combinations
* in a MACIP ACL.
*/
- vl_api_mac_address_t src_mac;
- vl_api_mac_address_t src_mac_mask;
- vl_api_prefix_t src_prefix;
+ u8 src_mac[6];
+ u8 src_mac_mask[6];
+ u8 src_ip_addr[16];
+ u8 src_ip_prefix_len;
};
+
+
#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;
- 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 (0, " %s ", a->is_ipv6 ? "ipv6" : "ipv4");
s = format_acl_action (s, a->is_permit);
s = format (s, " \\\n");
- s = format (s, " src %U dst %U \\\n",
- format_fib_prefix, &src,
- format_fib_prefix, &dst);
+ 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, " 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->src_prefix.address.af ? "ipv6" : "ipv4",
+ s = format (0, " %s %s \\\n", a->is_ipv6 ? "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);
- s = format (s, " src ip %U, \\",
- format_fib_prefix, &src);
+ 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);
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=0,
- d_prefix=0, d_ip=0):
- if ip:
- src_prefix = IPv6Network((s_ip, s_prefix))
- dst_prefix = IPv6Network((d_ip, d_prefix))
+ 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
else:
- 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()))
+ 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)))
# Apply a ACL on the interface as inbound
for i in self.pg_interfaces:
- 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()))
+ 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)))
# Apply a ACL on the interface as inbound
- acl_if = VppAclInterface(self, sw_if_index=sw_if_index, n_input=1,
- acls=[acl])
- return acl.acl_index
+ self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index,
+ n_input=1,
+ acls=[reply.acl_index])
+ return reply.acl_index
- def etype_whitelist(self, whitelist, n_input, add=True):
+ def etype_whitelist(self, whitelist, n_input):
# Apply whitelists on all the interfaces
- 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()
+ 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
def create_upper_layer(self, packet_index, proto, ports=0):
p = self.proto_map[proto]
"""
self.logger.info("ACLP_TEST_START_0001")
- # Create a permit-1234 ACL
- r = [AclRule(is_permit=1, proto=17, ports=1234)]
- r[0].sport_to = 1235
+ # 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}]
# Test 1: add a new ACL
- first_acl = VppAcl(self, rules=r, tag="permit 1234")
- first_acl.add_vpp_config()
- self.assertTrue(first_acl.query_vpp_config())
+ reply = self.vapi.acl_add_replace(acl_index=4294967295, r=r,
+ tag=b"permit 1234")
+ self.assertEqual(reply.retval, 0)
# The very first ACL gets #0
- self.assertEqual(first_acl.acl_index, 0)
- rr = first_acl.dump()
+ self.assertEqual(reply.acl_index, 0)
+ first_acl = reply.acl_index
+ rr = self.vapi.acl_dump(reply.acl_index)
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):
- encoded_rule = r[i_rule].encode()
- for rule_key in encoded_rule:
+ for rule_key in r[i_rule]:
self.assertEqual(rr[0].r[i_rule][rule_key],
- 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())
+ 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)
# The second ACL gets #1
- self.assertEqual(second_acl.acl_index, 1)
+ self.assertEqual(reply.acl_index, 1)
+ second_acl = reply.acl_index
# Test 2: try to modify a nonexistent ACL
- 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()
-
+ 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)
# 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=0,
- acls=[second_acl])
- acl_if_list.add_vpp_config()
- second_acl.remove_vpp_config(expect_error=True)
+ 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)
# Unapply an ACL and then try to delete it - must be ok
- acl_if_list.remove_vpp_config()
- second_acl.remove_vpp_config()
+ 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)
# try to apply a nonexistent ACL - must fail
- 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.vapi.acl_interface_set_acl_list(sw_if_index=self.pg0.sw_if_index,
+ n_input=1,
+ acls=[first_acl],
+ expected_retval=-6)
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, "permit per-flow")
+ acl_idx = self.apply_rules(rules, b"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, "deny per-flow;permit all")
+ acl_idx = self.apply_rules(rules, b"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, "permit icmpv4")
+ self.apply_rules(rules, b"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, "permit icmpv6")
+ self.apply_rules(rules, b"permit icmpv6")
# Traffic should still pass
self.run_verify_test(self.ICMP, self.IPV6,
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, "deny icmpv4")
+ self.apply_rules(rules, b"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, "deny icmpv6")
+ self.apply_rules(rules, b"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, "permit ipv4 tcp")
+ self.apply_rules(rules, b"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, "permit ip6 tcp")
+ self.apply_rules(rules, b"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, "permit ipv udp")
+ self.apply_rules(rules, b"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, "permit ip6 udp")
+ self.apply_rules(rules, b"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, "deny ip4/ip6 tcp")
+ self.apply_rules(rules, b"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, "deny ip4/ip6 udp")
+ self.apply_rules(rules, b"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]))
- acl = VppAcl(self, rules=rules)
- acl.add_vpp_config()
- result = acl.dump()
+ reply = self.vapi.acl_add_replace(acl_index=4294967295, r=rules)
+ result = self.vapi.acl_dump(reply.acl_index)
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, "permit ip4 tcp %d" % port)
+ self.apply_rules(rules, b"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, "permit ip4 tcp %d" % port)
+ self.apply_rules(rules, b"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, "permit ip4 tcp %d" % port)
+ self.apply_rules(rules, b"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 UDPv6
+ """ permit single UPPv6
"""
self.logger.info("ACLP_TEST_START_0018")
self.PORTS_ALL, 0))
# Apply rules
- self.apply_rules(rules, "permit ip4 tcp %d" % port)
+ self.apply_rules(rules, b"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, "deny ip4/ip6 udp %d" % port)
+ self.apply_rules(rules, b"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, "deny ip4/ip6 udp %d" % port)
+ self.apply_rules(rules, b"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, "deny ip4/ip6 udp %d" % port)
+ self.apply_rules(rules, b"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, "permit empty udp ip4 %d" % port)
+ self.apply_rules(rules, b"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, "permit empty udp ip6 %d" % port)
+ self.apply_rules(rules, b"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, "permit ipv4 tcp")
+ self.apply_rules(rules, b"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, "permit ip6 tcp")
+ self.apply_rules(rules, b"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, "permit ipv4 udp")
+ self.apply_rules(rules, b"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, "permit ip6 udp")
+ self.apply_rules(rules, b"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, "deny ip4/ip6 tcp")
+ self.apply_rules(rules, b"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, "deny ip4/ip6 udp")
+ self.apply_rules(rules, b"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, "permit ipv4 tcp")
+ self.apply_rules(rules, b"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, "permit ipv4 tcp")
+ self.apply_rules(rules, b"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, add=False)
+ self.etype_whitelist([], 0)
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, "permit ipv4 tcp")
+ self.apply_rules(rules, b"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, add=False)
+ self.etype_whitelist([], 0)
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, "permit ipv4 tcp")
+ self.apply_rules(rules, b"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, add=False)
+ self.etype_whitelist([], 0)
# 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, "permit ipv4 tcp", intf[0].sw_if_index)
+ self.apply_rules_to(rules, b"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 = 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
-
+ 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,
+ }
rules.append(new_rule)
- new_rule_permit = copy.copy(new_rule)
- new_rule_permit.is_permit = 1
+ new_rule_permit = new_rule.copy()
+ new_rule_permit['is_permit'] = 1
permit_rules.append(new_rule_permit)
- new_rule_permit_and_reflect = copy.copy(new_rule)
+ new_rule_permit_and_reflect = new_rule.copy()
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, acl_if):
- saved_n_input = acl_if.n_input
- # TOTO: maybe copy each one??
- saved_acls = acl_if.acls
+ 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])
# now create a list of all the rules in all ACLs
all_rules = []
- for old_acl in saved_acls:
- for rule in old_acl.rules:
- all_rules.append(rule)
+ for old_acl in orig_acls:
+ for rule in old_acl.r:
+ all_rules.append(dict(rule._asdict()))
# Add a few ACLs made from shuffled rules
shuffle(all_rules)
- acl1 = VppAcl(self, rules=all_rules[::2], tag="shuffle 1. acl")
- acl1.add_vpp_config()
-
+ reply = self.vapi.acl_add_replace(acl_index=4294967295,
+ r=all_rules[::2],
+ tag=b"shuffle 1. acl")
+ shuffle_acl_1 = reply.acl_index
shuffle(all_rules)
- acl2 = VppAcl(self, rules=all_rules[::3], tag="shuffle 2. acl")
- acl2.add_vpp_config()
-
+ reply = self.vapi.acl_add_replace(acl_index=4294967295,
+ r=all_rules[::3],
+ tag=b"shuffle 2. acl")
+ shuffle_acl_2 = reply.acl_index
shuffle(all_rules)
- acl3 = VppAcl(self, rules=all_rules[::2], tag="shuffle 3. acl")
- acl3.add_vpp_config()
+ reply = self.vapi.acl_add_replace(acl_index=4294967295,
+ r=all_rules[::2],
+ tag=b"shuffle 3. acl")
+ shuffle_acl_3 = reply.acl_index
# apply the shuffle ACLs in front
- input_acls = [acl1, acl2]
- output_acls = [acl1, acl2]
+ input_acls = [shuffle_acl_1, shuffle_acl_2]
+ output_acls = [shuffle_acl_1, shuffle_acl_2]
# add the currently applied ACLs
- n_input = acl_if.n_input
- input_acls.extend(saved_acls[:n_input])
- output_acls.extend(saved_acls[n_input:])
+ 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:])
# and the trailing shuffle ACL(s)
- input_acls.extend([acl3])
- output_acls.extend([acl3])
+ input_acls.extend([shuffle_acl_3])
+ output_acls.extend([shuffle_acl_3])
# set the interface ACL list to the result
- acl_if.n_input = len(input_acls)
- acl_if.acls = input_acls + output_acls
- acl_if.add_vpp_config()
-
+ self.vapi.acl_interface_set_acl_list(sw_if_index=sw_if_index,
+ n_input=len(input_acls),
+ acls=input_acls + output_acls)
# change the ACLs a few times
for i in range(1, 10):
shuffle(all_rules)
- acl1.rules = all_rules[::1+(i % 2)]
- acl1.add_vpp_config()
-
+ reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_1,
+ r=all_rules[::1+(i % 2)],
+ tag=b"shuffle 1. acl")
shuffle(all_rules)
- acl2.rules = all_rules[::1+(i % 3)]
- acl2.add_vpp_config()
-
+ reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_2,
+ r=all_rules[::1+(i % 3)],
+ tag=b"shuffle 2. acl")
shuffle(all_rules)
- acl3.rules = all_rules[::1+(i % 5)]
- acl3.add_vpp_config()
+ reply = self.vapi.acl_add_replace(acl_index=shuffle_acl_2,
+ r=all_rules[::1+(i % 5)],
+ tag=b"shuffle 3. acl")
# restore to how it was before and clean up
- 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()
+ 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)
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
- 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}
+ 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}
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
-
- 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)
+ 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)
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']
- 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)
+ 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)
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'].acl_index, pkts)
+ self.verify_acl_packet_count(acls['L3'], 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'].acl_index, pkts)
+ self.verify_acl_packet_count(acls['L2'], 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[7] = random.randint(100, 200)
+ ip6[8] = 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 = MacipRule(is_permit=self.PERMIT,
- src_prefix=ip_network((ip_pack, ip_len)),
- src_mac=MACAddress(mac).packed,
- src_mac_mask=MACAddress(mask).packed)
+ 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(':', ''))})
rules.append(rule)
if ip_type == self.WILD_IP:
break
return acls
def apply_macip_rules(self, acls):
- macip_acls = []
for acl in acls:
- macip_acl = VppMacipAcl(self, rules=acl)
- macip_acl.add_vpp_config()
- macip_acls.append(macip_acl)
- return macip_acls
+ reply = self.vapi.macip_acl_add(acl)
+ self.assertEqual(reply.retval, 0)
+ self.ACLS.append(reply.acl_index)
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
- 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_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}
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 = 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_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_rules.append(macip_rule)
# deny all other packets
if not (mac_type == self.WILD_MAC and ip_type == self.WILD_IP):
- 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_rule = ({'is_permit': 0,
+ 'is_ipv6': is_ip6,
+ 'src_ip_addr': "",
+ 'src_ip_prefix_len': 0,
+ 'src_mac': "",
+ 'src_mac_mask': ""})
macip_rules.append(macip_rule)
- 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_rule = {'is_permit': 0,
+ 'is_ipv6': is_ip6}
acl_rules.append(acl_rule)
return {'stream': packets,
'macip_rules': macip_rules,
if apply_rules:
if isMACIP:
- self.acl = VppMacipAcl(self, rules=test_dict['macip_rules'])
+ reply = self.vapi.macip_acl_add(test_dict['macip_rules'])
else:
- self.acl = VppAcl(self, rules=test_dict['acl_rules'])
- self.acl.add_vpp_config()
+ reply = self.vapi.acl_add_replace(acl_index=4294967295,
+ r=test_dict['acl_rules'])
+ self.assertEqual(reply.retval, 0)
+ acl_index = reply.acl_index
if isMACIP:
- 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)
+ 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])
else:
- 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()
+ self.vapi.acl_interface_add_del(
+ sw_if_index=tx_if.sw_if_index, acl_index=acl_index)
else:
- if hasattr(self, "acl_if"):
- self.acl_if.remove_vpp_config()
- if try_replace and hasattr(self, "acl"):
+ self.vapi.macip_acl_interface_add_del(
+ sw_if_index=tx_if.sw_if_index,
+ acl_index=0)
+ if try_replace:
if isMACIP:
- self.acl.rules = test_dict['macip_rules']
- self.acl.add_vpp_config()
+ reply = self.vapi.macip_acl_add_replace(
+ test_dict['macip_rules'],
+ acl_index)
else:
- self.acl.rules = test_dict['acl_rules']
- self.acl.add_vpp_config()
+ reply = self.vapi.acl_add_replace(acl_index=acl_index,
+ r=test_dict['acl_rules'])
+ self.assertEqual(reply.retval, 0)
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:
- if hasattr(self, "acl_if"):
- self.acl_if.remove_vpp_config()
- if hasattr(self, "acl"):
- self.acl.remove_vpp_config()
+ 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)
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)
- macip_acls = self.apply_macip_rules(r1)
+ self.apply_macip_rules(r1)
acls_before = self.macip_acl_dump_debug()
# replace acls #2, #3 with new
- macip_acls[2].rules = r2[0]
- macip_acls[2].add_vpp_config()
- macip_acls[3].rules = r2[1]
- macip_acls[3].add_vpp_config()
+ 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)
acls_after = self.macip_acl_dump_debug()
intf_count = len(self.interfaces)+1
intf = []
- macip_alcs = self.apply_macip_rules(
- self.create_rules(acl_count=3, rules_count=[3, 5, 4]))
+ 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
- macip_acl_if0 = VppMacipAclInterface(
- self, sw_if_index=sw_if_index0, acls=[macip_alcs[1]])
- macip_acl_if0.add_vpp_config()
+ self.vapi.macip_acl_interface_add_del(sw_if_index0, 1)
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
- macip_acl_if1 = VppMacipAclInterface(
- self, sw_if_index=sw_if_index1, acls=[macip_alcs[0]])
- macip_acl_if1.add_vpp_config()
+ self.vapi.macip_acl_interface_add_del(sw_if_index1, 0)
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
- 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()
+ self.vapi.macip_acl_interface_add_del(sw_if_index2, 1)
+ self.vapi.macip_acl_interface_add_del(sw_if_index3, 1)
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
#
- 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()
-
+ 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])
c1 = VppGbpContract(
- self, 400, epgs[0].sclass, epgs[1].sclass, acl.acl_index,
+ self, 400, epgs[0].sclass, epgs[1].sclass, 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.acl_index,
+ self, 400, epgs[1].sclass, epgs[0].sclass, 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.acl_index,
+ self, 400, epgs[0].sclass, epgs[2].sclass, 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, acl2.acl_index,
+ self, 400, epgs[0].sclass, epgs[3].sclass, acl_index2,
[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, acl2.acl_index,
+ self, 400, epgs[3].sclass, epgs[0].sclass, acl_index2,
[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
#
- 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()
-
+ 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])
c1 = VppGbpContract(
- self, 401, epg_220.sclass, epg_330.sclass, acl.acl_index,
+ self, 401, epg_220.sclass, epg_330.sclass, 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)
- 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()
-
+ 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])
c2 = VppGbpContract(
- self, 401, epg_330.sclass, epg_220.sclass, acl.acl_index,
+ self, 401, epg_330.sclass, epg_220.sclass, 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
#
- 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()
-
+ 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])
c1 = VppGbpContract(
- self, 400, epgs[0].sclass, epgs[1].sclass, acl.acl_index,
+ self, 400, epgs[0].sclass, epgs[1].sclass, 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.acl_index,
+ self, 400, epgs[1].sclass, epgs[0].sclass, 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.acl_index,
+ self, 400, epgs[0].sclass, epgs[2].sclass, 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
#
- 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()
+ 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])
#
# test the src-ip hash mode
#
c1 = VppGbpContract(
- self, 402, epg_220.sclass, epg_222.sclass, acl.acl_index,
+ self, 402, epg_220.sclass, epg_222.sclass, 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.acl_index,
+ self, 402, epg_222.sclass, epg_220.sclass, 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.acl_index,
+ self, 402, epg_220.sclass, epg_222.sclass, 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.acl_index,
+ self, 402, epg_222.sclass, epg_220.sclass, 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.acl_index,
+ self, 402, epg_220.sclass, epg_221.sclass, 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.acl_index,
+ self, 402, epg_221.sclass, epg_220.sclass, 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.acl_index,
+ self, 402, epg_220.sclass, epg_221.sclass, 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.acl_index,
+ self, 402, 4220, 4221, 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.acl_index,
+ self, 402, 4220, 4221, 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.acl_index,
+ self, 402, epg_221.sclass, epg_222.sclass, 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
#
- 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()
+ 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])
#
# test the src-ip hash mode
#
c1 = VppGbpContract(
- self, 402, epg_220.sclass, epg_221.sclass, acl.acl_index,
+ self, 402, epg_220.sclass, epg_221.sclass, 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.acl_index,
+ self, 402, epg_221.sclass, epg_220.sclass, 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.acl_index,
+ self, 402, epg_320.sclass, epg_221.sclass, 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
#
- 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()
+ 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])
#
# A contract with the wrong scope is not matched
#
c_44 = VppGbpContract(
- self, 44, 4220, 4221, acl.acl_index,
+ self, 44, 4220, 4221, 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.acl_index,
+ self, 55, 4220, 4221, 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.acl_index,
+ self, 55, 4220, 113, 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.acl_index,
+ self, 55, 113, 4220, 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.acl_index,
+ self, 55, 4220, 4222, 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
#
- 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()
+ 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])
c1 = VppGbpContract(
- self, 55, 4220, 4221, acl.acl_index,
+ self, 55, 4220, 4221, 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.acl_index,
+ self, 55, 4220, 113, 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.acl_index,
+ self, 55, 113, 4220, 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.acl_index,
+ self, 55, 4220, 4222, 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 = 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()
+ 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
# apply as input acl on interface and confirm it blocks everything
- acl_if = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index,
- n_input=1, acls=[out2in_acl])
- acl_if.add_vpp_config()
+ self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
+ n_input=1,
+ acls=[out2in_acl])
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
- acl_if.acls = [out2in_acl, in2out_acl]
- acl_if.add_vpp_config()
+ self.vapi.acl_interface_set_acl_list(sw_if_index=self.pg1.sw_if_index,
+ n_input=1,
+ acls=[out2in_acl, in2out_acl])
# 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 = 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()
+ 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])
#
# Apply the ACL on the output interface
#
- acl_if1 = VppAclInterface(self, sw_if_index=self.pg1.sw_if_index,
- n_input=0, acls=[acl])
- acl_if1.add_vpp_config()
+ self.vapi.acl_interface_set_acl_list(self.pg1.sw_if_index,
+ 0,
+ [acl.acl_index])
#
# Send packet's that should match the ACL and be dropped
#
# cleanup
#
- acl_if1.remove_vpp_config()
- acl.remove_vpp_config()
+ self.vapi.acl_interface_set_acl_list(self.pg1.sw_if_index,
+ 0, [])
+ self.vapi.acl_del(acl.acl_index)
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 = 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()
+ 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])
# Apply the ACL on the pipe on output
- acl_if_e = VppAclInterface(self, sw_if_index=pipes[0].east, n_input=0,
- acls=[acl])
- acl_if_e.add_vpp_config()
-
+ self.vapi.acl_interface_set_acl_list(pipes[0].east,
+ 0,
+ [acl.acl_index])
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
- 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.vapi.acl_interface_set_acl_list(pipes[0].east,
+ 0,
+ [])
+ self.vapi.acl_interface_set_acl_list(pipes[0].west,
+ 1,
+ [acl.acl_index])
self.send_and_assert_no_replies(self.pg0, p * NUM_PKTS)
self.send_and_expect(self.pg1, p * NUM_PKTS, self.pg0)
-
- acl_if_w.remove_vpp_config()
+ self.vapi.acl_interface_set_acl_list(pipes[0].west,
+ 0,
+ [])
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
#
- acl_if_e1 = VppAclInterface(self, sw_if_index=pipes[1].east, n_input=0,
- acls=[acl])
- acl_if_e1.add_vpp_config()
+ self.vapi.acl_interface_set_acl_list(pipes[1].east,
+ 0,
+ [acl.acl_index])
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
- 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.vapi.acl_interface_set_acl_list(pipes[1].east,
+ 0,
+ [])
+ self.vapi.acl_interface_set_acl_list(pipes[1].west,
+ 1,
+ [acl.acl_index])
self.send_and_assert_no_replies(self.pg2, p_east * NUM_PKTS)
self.send_and_expect(self.pg3, p_west * NUM_PKTS, self.pg2)
- acl_if_w1.remove_vpp_config()
-
+ self.vapi.acl_interface_set_acl_list(pipes[1].west,
+ 0,
+ [])
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