Change the definition of vl_api_mac_address_t to an aliased type.
Change-Id: I1434f316d0fad6a099592f39bceeb8faeaf1d134
Signed-off-by: Ole Troan <ot@cisco.com>
return addr;
}
-vapi_type_mac_address
-to_api(const mac_address_t& a)
+void
+to_api(const mac_address_t& a, vapi_type_mac_address& v)
{
- vapi_type_mac_address v;
-
- std::copy(std::begin(a.bytes), std::end(a.bytes), v.bytes);
-
- return (v);
+ std::copy(std::begin(a.bytes), std::end(a.bytes), v);
}
mac_address_t
from_api(const vapi_type_mac_address& v)
{
- return mac_address_t(v.bytes);
+ return mac_address_t(v);
}
route::prefix_t
ip_address_t from_api(const vapi_type_address& v);
ip_address_t from_api(const vapi_type_ip4_address& v);
-vapi_type_mac_address to_api(const mac_address_t& a);
+void to_api(const mac_address_t& a, vapi_type_mac_address& m);
mac_address_t from_api(const vapi_type_mac_address& v);
auto& payload = req.get_request().get_payload();
payload.bd_id = m_bd;
payload.is_add = 1;
- payload.mac = to_api(m_mac);
+ to_api(m_mac, payload.mac);
to_api(m_ip_addr, payload.ip);
VAPI_CALL(req.execute());
auto& payload = req.get_request().get_payload();
payload.bd_id = m_bd;
payload.is_add = 0;
- payload.mac = to_api(m_mac);
+ to_api(m_mac, payload.mac);
to_api(m_ip_addr, payload.ip);
VAPI_CALL(req.execute());
payload.contract.rules[ii].nh_set.n_nhs = nh_size;
while (jj < nh_size) {
to_api(nh_it->getIp(), payload.contract.rules[ii].nh_set.nhs[jj].ip);
- payload.contract.rules[ii].nh_set.nhs[jj].mac = to_api(nh_it->getMac());
+ to_api(nh_it->getMac(), payload.contract.rules[ii].nh_set.nhs[jj].mac);
payload.contract.rules[ii].nh_set.nhs[jj].bd_id = nh_it->getBdId();
payload.contract.rules[ii].nh_set.nhs[jj].rd_id = nh_it->getRdId();
++nh_it;
for (n = 0; n < payload.endpoint.n_ips; n++) {
to_api(m_ip_addrs[n], payload.endpoint.ips[n]);
}
- payload.endpoint.mac = to_api(m_mac);
+ to_api(m_mac, payload.endpoint.mac);
VAPI_CALL(req.execute());
ip_address_decode (&mp->endpoint.ips[ii], &ips[ii]);
}
}
- mac_address_decode (&mp->endpoint.mac, &mac);
+ mac_address_decode (mp->endpoint.mac, &mac);
if (GBP_ENDPOINT_FLAG_REMOTE & gef)
{
mp->endpoint.flags = gbp_endpoint_flags_encode (gef->gef_flags);
mp->handle = htonl (gei);
mp->age = vlib_time_now (vlib_get_main ()) - ge->ge_last_time;
- mac_address_encode (&ge->ge_key.gek_mac, &mp->endpoint.mac);
+ mac_address_encode (&ge->ge_key.gek_mac, mp->endpoint.mac);
vec_foreach_index (ii, ge->ge_key.gek_ips)
{
return (VNET_API_ERROR_NO_SUCH_FIB);
ip_address_decode (&in->ip, &ip);
- mac_address_decode (&in->mac, &mac);
+ mac_address_decode (in->mac, &mac);
*gnhi = gbp_next_hop_alloc (&ip, grd, &mac, gbd);
#if VPP_API_TEST_BUILTIN == 0
#include <netdb.h>
-/* *INDENT-OFF* */
-const mac_address_t ZERO_MAC_ADDRESS = {
- .bytes = {
- 0, 0, 0, 0, 0, 0,
- },
-};
-/* *INDENT-ON* */
-
u32
vl (void *p)
{
api_bd_ip_mac_add_del (vat_main_t * vam)
{
vl_api_address_t ip = VL_API_ZERO_ADDRESS;
- vl_api_mac_address_t mac = VL_API_ZERO_MAC_ADDRESS;
+ vl_api_mac_address_t mac = { 0 };
unformat_input_t *i = vam->input;
vl_api_bd_ip_mac_add_del_t *mp;
ip46_type_t type;
* limitations under the License.
*/
-typedef mac_address {
- u8 bytes[6];
-};
+typedef u8 mac_address[6];
#undef vl_printfun
void
-mac_address_decode (const vl_api_mac_address_t * in, mac_address_t * out)
+mac_address_decode (const u8 * in, mac_address_t * out)
{
- mac_address_from_bytes (out, in->bytes);
+ mac_address_from_bytes (out, in);
}
void
-mac_address_encode (const mac_address_t * in, vl_api_mac_address_t * out)
+mac_address_encode (const mac_address_t * in, u8 * out)
{
- clib_memcpy_fast (out->bytes, in->bytes, 6);
+ clib_memcpy_fast (out, in->bytes, 6);
}
/*
#include <vnet/ethernet/mac_address.h>
-/**
- * Forward declarations so we need not #include the API definitions here
- */
-struct _vl_api_mac_address;
-
-extern void mac_address_decode (const struct _vl_api_mac_address *in,
- mac_address_t * out);
-extern void mac_address_encode (const mac_address_t * in,
- struct _vl_api_mac_address *out);
+extern void mac_address_decode (const u8 * in, mac_address_t * out);
+extern void mac_address_encode (const mac_address_t * in, u8 * out);
#endif
u32 bd_id;
u8 is_ipv6;
u8 ip_address[16];
- u8 mac_address[6];
+ vl_api_mac_address_t mac_address;
};
/** \brief Dump bridge domain IP to MAC entries
bd_index = p[0];
type = ip_address_decode (&mp->ip, &ip_addr);
- mac_address_decode (&mp->mac, &mac);
+ mac_address_decode (mp->mac, &mac);
if (bd_add_del_ip_mac (bd_index, type, &ip_addr, &mac, mp->is_add))
rv = VNET_API_ERROR_UNSPECIFIED;
--- /dev/null
+#!/usr/bin/env python
+#
+# Copyright (c) 2016 Cisco and/or its affiliates.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import binascii
+
+
+def mac_pton(s):
+ '''Convert MAC address as text to binary'''
+ return binascii.unhexlify(s.replace(':', ''))
+
+
+def mac_ntop(binary):
+ '''Convert MAC address as binary to text'''
+ x = b':'.join(binascii.hexlify(binary)[i:i + 2]
+ for i in range(0, 12, 2))
+ return str(x.decode('ascii'))
+
+
+class MACAddress():
+ def __init__(self, mac):
+ '''MAC Address as a text-string (aa:bb:cc:dd:ee:ff) or 6 bytes'''
+ # Of course Python 2 doesn't distinguish str from bytes
+ if type(mac) is bytes and len(mac) == 6:
+ self.mac_binary = mac
+ self.mac_string = mac_ntop(mac)
+ else:
+ self.mac_binary = mac_pton(mac)
+ self.mac_string = mac
+
+ @property
+ def packed(self):
+ return self.mac_binary
+
+ def __len__(self):
+ return 6
+
+ def __str__(self):
+ return self.mac_string
+
+ def __repr__(self):
+ return '%s(%s)' % (self.__class__.__name__, self.mac_string)
from socket import inet_pton, inet_ntop, AF_INET6, AF_INET
import socket
import ipaddress
+from . import macaddress
# Copies from vl_api_address_t definition
ADDRESS_IP4 = 0
'len': o.prefixlen},
'str': lambda s: format_vl_api_prefix_t(s)
},
+ 'vl_api_mac_address_t':
+ {
+ 'MACAddress': lambda o: o.packed,
+ 'str': lambda s: macaddress.mac_pton(s)
+ },
}
'vl_api_ip4_prefix_t': lambda o: ipaddress.IPv4Network((o.prefix, o.len)),
'vl_api_address_t': lambda o: unformat_api_address_t(o),
'vl_api_prefix_t': lambda o: unformat_api_prefix_t(o),
+ 'vl_api_mac_address_t': lambda o: macaddress.MACAddress(o),
}
import atexit
from . vpp_serializer import VPPType, VPPEnumType, VPPUnionType, BaseTypes
from . vpp_serializer import VPPMessage, vpp_get_type, VPPTypeAlias
+from . macaddress import MACAddress, mac_pton, mac_ntop
logger = logging.getLogger(__name__)
vpp_instance.logger.debug('Cleaning up VPP on exit')
vpp_instance.disconnect()
+
if sys.version[0] == '2':
def vpp_iterator(d):
return d.iteritems()
logger = logging.getLogger(__name__)
if sys.version[0] == '2':
- check = lambda d: type(d) is dict
+ def check(d): type(d) is dict
else:
- check = lambda d: type(d) is dict or type(d) is bytes
+ def check(d): type(d) is dict or type(d) is bytes
+
def conversion_required(data, field_type):
if check(data):
return b'', 0
p = BaseTypes('u8', length)
x, size = p.unpack(data, offset + length_field_size)
- x2 = x.split(b'\0',1)[0]
+ x2 = x.split(b'\0', 1)[0]
return (x2.decode('utf8'), size + length_field_size)
{
vl_api_mac_address_t *mac = va_arg (*args, vl_api_mac_address_t *);
- return (unformat (input, "%U",unformat_ethernet_address, mac->bytes));
+ return (unformat (input, "%U",unformat_ethernet_address, mac));
}
uword
{
vl_api_mac_address_t *mac = va_arg (*args, vl_api_mac_address_t *);
- return (format (s, "%U", format_ethernet_address, mac->bytes));
+ return (format (s, "%U", format_ethernet_address, mac));
}
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
-from vpp_mac import mactobinary
from vpp_bond_interface import VppBondInterface
+from vpp_papi import MACAddress
class TestBondInterface(VppTestCase):
# create interface (BondEthernet0)
# self.logger.info("create bond")
bond0_mac = "02:fe:38:30:59:3c"
- mac = mactobinary(bond0_mac)
+ mac = MACAddress(bond0_mac).packed
bond0 = VppBondInterface(self,
mode=3,
lb=1,
from vpp_neighbor import VppNeighbor
from vpp_ip_route import find_route, VppIpTable
from util import mk_ll_addr
-from vpp_mac import mactobinary, binarytomac
from scapy.layers.l2 import Ether, getmacbyip, ARP
from scapy.layers.inet import IP, UDP, ICMP
from scapy.layers.inet6 import IPv6, in6_getnsmac
from socket import AF_INET, AF_INET6
from scapy.utils import inet_pton, inet_ntop
from scapy.utils6 import in6_ptop
+from vpp_papi import mac_pton
DHCP4_CLIENT_PORT = 68
DHCP4_SERVER_PORT = 67
UDP(sport=DHCP4_SERVER_PORT, dport=DHCP4_CLIENT_PORT) /
BOOTP(op=1,
yiaddr=self.pg3.local_ip4,
- chaddr=mactobinary(self.pg3.local_mac)) /
+ chaddr=mac_pton(self.pg3.local_mac)) /
DHCP(options=[('message-type', 'offer'),
('server_id', self.pg3.remote_ip4),
'end']))
IP(src=self.pg3.remote_ip4, dst="255.255.255.255") /
UDP(sport=DHCP4_SERVER_PORT, dport=DHCP4_CLIENT_PORT) /
BOOTP(op=1, yiaddr=self.pg3.local_ip4,
- chaddr=mactobinary(self.pg3.local_mac)) /
+ chaddr=mac_pton(self.pg3.local_mac)) /
DHCP(options=[('message-type', 'ack'),
('subnet_mask', "255.255.255.0"),
('router', self.pg3.remote_ip4),
# remove the left over ARP entry
self.vapi.ip_neighbor_add_del(self.pg3.sw_if_index,
- mactobinary(self.pg3.remote_mac),
+ mac_pton(self.pg3.remote_mac),
self.pg3.remote_ip4,
is_add=0)
#
IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
UDP(sport=DHCP4_SERVER_PORT, dport=DHCP4_CLIENT_PORT) /
BOOTP(op=1, yiaddr=self.pg3.local_ip4,
- chaddr=mactobinary(self.pg3.local_mac)) /
+ chaddr=mac_pton(self.pg3.local_mac)) /
DHCP(options=[('message-type', 'ack'),
('subnet_mask', "255.255.255.0"),
('router', self.pg3.remote_ip4),
IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
UDP(sport=DHCP4_SERVER_PORT, dport=DHCP4_CLIENT_PORT) /
BOOTP(op=1, yiaddr=self.pg3.local_ip4,
- chaddr=mactobinary(self.pg3.local_mac)) /
+ chaddr=mac_pton(self.pg3.local_mac)) /
DHCP(options=[('message-type', 'offer'),
('server_id', self.pg3.remote_ip4),
'end']))
IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
UDP(sport=DHCP4_SERVER_PORT, dport=DHCP4_CLIENT_PORT) /
BOOTP(op=1, yiaddr=self.pg3.local_ip4,
- chaddr=mactobinary(self.pg3.local_mac)) /
+ chaddr=mac_pton(self.pg3.local_mac)) /
DHCP(options=[('message-type', 'ack'),
('subnet_mask', "255.255.255.0"),
('router', self.pg3.remote_ip4),
# remove the left over ARP entry
self.vapi.ip_neighbor_add_del(self.pg3.sw_if_index,
- mactobinary(self.pg3.remote_mac),
+ mac_pton(self.pg3.remote_mac),
self.pg3.remote_ip4,
is_add=0)
UDP(sport=DHCP4_SERVER_PORT, dport=DHCP4_CLIENT_PORT) /
BOOTP(op=1,
yiaddr=self.pg3.local_ip4,
- chaddr=mactobinary(self.pg3.local_mac)) /
+ chaddr=mac_pton(self.pg3.local_mac)) /
DHCP(options=[('message-type', 'offer'),
('server_id', self.pg3.remote_ip4),
('lease_time', lease_time),
IP(src=self.pg3.remote_ip4, dst='255.255.255.255') /
UDP(sport=DHCP4_SERVER_PORT, dport=DHCP4_CLIENT_PORT) /
BOOTP(op=1, yiaddr=self.pg3.local_ip4,
- chaddr=mactobinary(self.pg3.local_mac)) /
+ chaddr=mac_pton(self.pg3.local_mac)) /
DHCP(options=[('message-type', 'ack'),
('subnet_mask', '255.255.255.0'),
('router', self.pg3.remote_ip4),
# remove the left over ARP entry
self.vapi.ip_neighbor_add_del(self.pg3.sw_if_index,
- mactobinary(self.pg3.remote_mac),
+ mac_pton(self.pg3.remote_mac),
self.pg3.remote_ip4,
is_add=0)
from vpp_sub_interface import VppDot1QSubint
from vpp_ip import *
-from vpp_mac import *
from vpp_papi_provider import L2_PORT_TYPE
-from vpp_papi import VppEnum
+from vpp_papi import VppEnum, MACAddress
from scapy.packet import Raw
from scapy.layers.l2 import Ether, ARP, Dot1Q
if ip:
vip = VppIpAddress(ip)
if mac:
- vmac = VppMacAddress(mac)
+ vmac = MACAddress(mac)
eps = test.vapi.gbp_endpoint_dump()
if vip == eip:
return True
if mac:
- if vmac == ep.endpoint.mac:
+ if vmac.packed == ep.endpoint.mac:
return True
return False
GBP Endpoint
"""
- @property
- def bin_mac(self):
- return self.vmac.bytes
-
@property
def mac(self):
- return self.vmac.address
+ return str(self.vmac)
@property
def mac(self):
self._fip6 = VppIpAddress(fip6)
if mac:
- self.vmac = VppMacAddress(self.itf.remote_mac)
+ self.vmac = MACAddress(self.itf.remote_mac)
else:
- self.vmac = VppMacAddress("00:00:00:00:00:00")
+ self.vmac = MACAddress("00:00:00:00:00:00")
self.flags = flags
self.tun_src = VppIpAddress(tun_src)
res = self._test.vapi.gbp_endpoint_add(
self.itf.sw_if_index,
[self.ip4.encode(), self.ip6.encode()],
- self.vmac.encode(),
+ self.vmac.packed,
self.epg.epg,
self.flags,
self.tun_src.encode(),
def encode(self):
return {'ip': self.ip.encode(),
- 'mac': self.mac.encode(),
+ 'mac': self.mac.packed,
'bd_id': self.bd.bd.bd_id,
'rd_id': self.rd.rd_id}
self.create_pg_interfaces(range(9))
self.create_loopback_interfaces(8)
- self.router_mac = VppMacAddress("00:11:22:33:44:55")
+ self.router_mac = MACAddress("00:11:22:33:44:55")
for i in self.pg_interfaces:
i.admin_up()
rx = self.send_and_expect(src, tx, dst)
for r in rx:
- self.assertEqual(r[Ether].src, self.router_mac.address)
+ self.assertEqual(r[Ether].src, str(self.router_mac))
self.assertEqual(r[Ether].dst, dst.remote_mac)
self.assertEqual(r[IP].dst, dst_ip)
self.assertEqual(r[IP].src, src_ip)
rx = self.send_and_expect(src, tx, dst)
for r in rx:
- self.assertEqual(r[Ether].src, self.router_mac.address)
+ self.assertEqual(r[Ether].src, str(self.router_mac))
self.assertEqual(r[Ether].dst, dst.remote_mac)
self.assertEqual(r[IPv6].dst, dst_ip)
self.assertEqual(r[IPv6].src, src_ip)
VppIpInterfaceBind(self, epg.bvi, epg.rd.t6).add_vpp_config()
self.vapi.sw_interface_set_mac_address(
epg.bvi.sw_if_index,
- self.router_mac.bytes)
+ self.router_mac.packed)
# The BVIs are NAT inside interfaces
self.vapi.nat44_interface_add_del_feature(epg.bvi.sw_if_index,
# add the BD ARP termination entry for BVI IP
epg.bd_arp_ip4 = VppBridgeDomainArpEntry(self, epg.bd.bd,
- self.router_mac.address,
+ str(self.router_mac),
epg.bvi_ip4)
epg.bd_arp_ip6 = VppBridgeDomainArpEntry(self, epg.bd.bd,
- self.router_mac.address,
+ str(self.router_mac),
epg.bvi_ip6)
epg.bd_arp_ip4.add_vpp_config()
epg.bd_arp_ip6.add_vpp_config()
# packets to non-local L3 destinations dropped
#
pkt_intra_epg_220_ip4 = (Ether(src=self.pg0.remote_mac,
- dst=self.router_mac.address) /
+ dst=str(self.router_mac)) /
IP(src=eps[0].ip4.address,
dst="10.0.0.99") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
pkt_inter_epg_222_ip4 = (Ether(src=self.pg0.remote_mac,
- dst=self.router_mac.address) /
+ dst=str(self.router_mac)) /
IP(src=eps[0].ip4.address,
dst="10.0.1.99") /
UDP(sport=1234, dport=1234) /
self.send_and_assert_no_replies(self.pg0, pkt_intra_epg_220_ip4 * 65)
pkt_inter_epg_222_ip6 = (Ether(src=self.pg0.remote_mac,
- dst=self.router_mac.address) /
+ dst=str(self.router_mac)) /
IPv6(src=eps[0].ip6.address,
dst="2001:10::99") /
UDP(sport=1234, dport=1234) /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
pkt_inter_epg_220_to_222 = (Ether(src=self.pg0.remote_mac,
- dst=self.router_mac.address) /
+ dst=str(self.router_mac)) /
IP(src=eps[0].ip4.address,
dst=eps[3].ip4.address) /
UDP(sport=1234, dport=1234) /
self.send_and_expect_routed(eps[0].itf,
pkt_inter_epg_220_to_222 * 65,
eps[3].itf,
- self.router_mac.address)
+ str(self.router_mac))
#
# remove both contracts, traffic stops in both directions
# From an EP to an outside addess: IN2OUT
#
pkt_inter_epg_220_to_global = (Ether(src=self.pg0.remote_mac,
- dst=self.router_mac.address) /
+ dst=str(self.router_mac)) /
IP(src=eps[0].ip4.address,
dst="1.1.1.1") /
UDP(sport=1234, dport=1234) /
eps[0].fip4.address)
pkt_inter_epg_220_to_global = (Ether(src=self.pg0.remote_mac,
- dst=self.router_mac.address) /
+ dst=str(self.router_mac)) /
IPv6(src=eps[0].ip6.address,
dst="6001::1") /
UDP(sport=1234, dport=1234) /
#
# From a global address to an EP: OUT2IN
#
- pkt_inter_epg_220_from_global = (Ether(src=self.router_mac.address,
+ pkt_inter_epg_220_from_global = (Ether(src=str(self.router_mac),
dst=self.pg0.remote_mac) /
IP(dst=eps[0].fip4.address,
src="1.1.1.1") /
eps[0].itf,
eps[0].ip4.address)
- pkt_inter_epg_220_from_global = (Ether(src=self.router_mac.address,
+ pkt_inter_epg_220_from_global = (Ether(src=str(self.router_mac),
dst=self.pg0.remote_mac) /
IPv6(dst=eps[0].fip6.address,
src="6001::1") /
# IN2OUT2IN
#
pkt_intra_epg_220_global = (Ether(src=self.pg0.remote_mac,
- dst=self.router_mac.address) /
+ dst=str(self.router_mac)) /
IP(src=eps[0].ip4.address,
dst=eps[1].fip4.address) /
UDP(sport=1234, dport=1234) /
eps[1].ip4.address)
pkt_intra_epg_220_global = (Ether(src=self.pg0.remote_mac,
- dst=self.router_mac.address) /
+ dst=str(self.router_mac)) /
IPv6(src=eps[0].ip6.address,
dst=eps[1].fip6.address) /
UDP(sport=1234, dport=1234) /
rd1 = VppGbpRouteDomain(self, 2, t4, t6, tun_ip4_uu, tun_ip6_uu)
rd1.add_vpp_config()
- self.loop0.set_mac(self.router_mac.address)
+ self.loop0.set_mac(self.router_mac)
#
# Bind the BVI to the RD
rd1 = VppGbpRouteDomain(self, 2, t4, t6)
rd1.add_vpp_config()
- self.loop0.set_mac(self.router_mac.address)
+ self.loop0.set_mac(self.router_mac)
#
# Bind the BVI to the RD
# an L3 switch packet between local EPs in different EPGs
# different dest ports on each so the are LB hashed differently
#
- p4 = [(Ether(src=ep1.mac, dst=self.router_mac.address) /
+ p4 = [(Ether(src=ep1.mac, dst=str(self.router_mac)) /
IP(src=ep1.ip4.address, dst=ep2.ip4.address) /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100)),
- (Ether(src=ep2.mac, dst=self.router_mac.address) /
+ (Ether(src=ep2.mac, dst=str(self.router_mac)) /
IP(src=ep2.ip4.address, dst=ep1.ip4.address) /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))]
- p6 = [(Ether(src=ep1.mac, dst=self.router_mac.address) /
+ p6 = [(Ether(src=ep1.mac, dst=str(self.router_mac)) /
IPv6(src=ep1.ip6.address, dst=ep2.ip6.address) /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100)),
- (Ether(src=ep2.mac, dst=self.router_mac.address) /
+ (Ether(src=ep2.mac, dst=str(self.router_mac)) /
IPv6(src=ep2.ip6.address, dst=ep1.ip6.address) /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))]
dst=self.pg7.local_ip4) /
UDP(sport=1234, dport=48879) /
VXLAN(vni=444, gpid=221, flags=0x88) /
- Ether(src="00:22:22:22:22:33", dst=self.router_mac.address) /
+ Ether(src="00:22:22:22:22:33", dst=str(self.router_mac)) /
IP(src="10.0.0.88", dst=ep1.ip4.address) /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
dst=self.pg7.local_ip4) /
UDP(sport=1234, dport=48879) /
VXLAN(vni=444, gpid=221, flags=0x88) /
- Ether(src="00:22:22:22:22:33", dst=self.router_mac.address) /
+ Ether(src="00:22:22:22:22:33", dst=str(self.router_mac)) /
IPv6(src="2001:10::88", dst=ep1.ip6.address) /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
#
# L3 switch from local to remote EP
#
- p4 = [(Ether(src=ep1.mac, dst=self.router_mac.address) /
+ p4 = [(Ether(src=ep1.mac, dst=str(self.router_mac)) /
IP(src=ep1.ip4.address, dst="10.0.0.88") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))]
- p6 = [(Ether(src=ep1.mac, dst=self.router_mac.address) /
+ p6 = [(Ether(src=ep1.mac, dst=str(self.router_mac)) /
IPv6(src=ep1.ip6.address, dst="2001:10::88") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))]
rd1 = VppGbpRouteDomain(self, 2, t4, t6)
rd1.add_vpp_config()
- self.loop0.set_mac(self.router_mac.address)
+ self.loop0.set_mac(self.router_mac)
#
# Bind the BVI to the RD
# packets destined to unkown addresses in the BVI's subnet
# are ARP'd for
#
- p4 = (Ether(src=self.pg0.remote_mac, dst=self.router_mac.address) /
+ p4 = (Ether(src=self.pg0.remote_mac, dst=str(self.router_mac)) /
Dot1Q(vlan=100) /
IP(src="10.0.0.1", dst="10.0.0.88") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
- p6 = (Ether(src=self.pg0.remote_mac, dst=self.router_mac.address) /
+ p6 = (Ether(src=self.pg0.remote_mac, dst=str(self.router_mac)) /
Dot1Q(vlan=100) /
IPv6(src="2001:10::1", dst="2001:10::88") /
UDP(sport=1234, dport=1234) /
dst=self.pg7.local_ip4) /
UDP(sport=1234, dport=48879) /
VXLAN(vni=444, gpid=220, flags=0x88) /
- Ether(src=self.pg0.remote_mac, dst=self.router_mac.address) /
+ Ether(src=self.pg0.remote_mac, dst=str(self.router_mac)) /
IP(src="10.0.0.101", dst="10.0.0.1") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
dst=self.pg7.local_ip4) /
UDP(sport=1234, dport=48879) /
VXLAN(vni=444, gpid=220, flags=0x88) /
- Ether(src=self.pg0.remote_mac, dst=self.router_mac.address) /
+ Ether(src=self.pg0.remote_mac, dst=str(self.router_mac)) /
IP(src="10.0.0.101", dst="10.220.0.1") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
dst=self.pg7.local_ip4) /
UDP(sport=1234, dport=48879) /
VXLAN(vni=444, gpid=220, flags=0x88) /
- Ether(src=self.pg0.remote_mac, dst=self.router_mac.address) /
+ Ether(src=self.pg0.remote_mac, dst=str(self.router_mac)) /
IP(src="10.0.0.101", dst="10.200.0.1") /
UDP(sport=1234, dport=1234) /
Raw('\xa5' * 100))
#
# from the the subnet in EPG 220 beyond the external to remote
#
- p4 = (Ether(src=self.pg0.remote_mac, dst=self.router_mac.address) /
+ p4 = (Ether(src=self.pg0.remote_mac, dst=str(self.router_mac)) /
Dot1Q(vlan=100) /
IP(src="10.220.0.1", dst=rep.ip4.address) /
UDP(sport=1234, dport=1234) /
# from the the subnet in EPG 200 beyond the external to remote
# dropped due to no contract
#
- p4 = (Ether(src=self.pg0.remote_mac, dst=self.router_mac.address) /
+ p4 = (Ether(src=self.pg0.remote_mac, dst=str(self.router_mac)) /
Dot1Q(vlan=100) /
IP(src="10.200.0.1", dst=rep.ip4.address) /
UDP(sport=1234, dport=1234) /
from scapy.layers.inet import IP, UDP
from framework import VppTestCase, VppTestRunner
-from vpp_mac import mactobinary
+from vpp_papi import MACAddress
from vpp_papi_provider import L2_PORT_TYPE
self.send_and_verify_l2_to_ip()
# change the BVI's mac and resed traffic
- self.loop0.set_mac("00:00:00:11:11:33")
+ self.loop0.set_mac(MACAddress("00:00:00:11:11:33"))
self.send_and_verify_l2_to_ip()
# check it wasn't flooded
from framework import VppTestCase, VppTestRunner
from util import Host, ppp
+from vpp_papi import mac_pton
# from src/vnet/l2/l2_fib.h
MAC_EVENT_ACTION_ADD = 0
swif = pg_if.sw_if_index
for host in hosts[swif]:
self.vapi.l2fib_add_del(
- host.mac, bd_id, swif, static_mac=1)
+ mac_pton(host.mac), bd_id, swif, static_mac=1)
def delete_l2_fib_entry(self, bd_id, hosts):
"""
swif = pg_if.sw_if_index
for host in hosts[swif]:
self.vapi.l2fib_add_del(
- host.mac, bd_id, swif, is_add=0)
+ mac_pton(host.mac), bd_id, swif, is_add=0)
def flush_int(self, swif, learned_hosts):
"""
from framework import VppTestCase, VppTestRunner
from util import Host, ppp
-from vpp_mac import VppMacAddress, mactobinary
class TestL2bdArpTerm(VppTestCase):
for e in entries:
ip = e.ip4 if is_ipv6 == 0 else e.ip6
self.vapi.bd_ip_mac_add_del(bd_id=bd_id,
- mac=VppMacAddress(e.mac).encode(),
+ mac=e.mac,
ip=ip,
is_ipv6=is_ipv6,
is_add=is_add)
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from time import sleep
from util import ip4_range
-from vpp_mac import mactobinary
+from vpp_papi import mac_pton
from syslog_rfc5424_parser import SyslogMessage, ParseError
from syslog_rfc5424_parser.constants import SyslogFacility, SyslogSeverity
from vpp_papi_provider import SYSLOG_SEVERITY
""" NAT44 interfaces without configured IP address """
self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- mactobinary(self.pg7.remote_mac),
+ mac_pton(self.pg7.remote_mac),
self.pg7.remote_ip4n,
is_static=1)
self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
- mactobinary(self.pg8.remote_mac),
+ mac_pton(self.pg8.remote_mac),
self.pg8.remote_ip4n,
is_static=1)
""" NAT44 interfaces without configured IP address - 1:1 NAT """
self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- mactobinary(self.pg7.remote_mac),
+ mac_pton(self.pg7.remote_mac),
self.pg7.remote_ip4n,
is_static=1)
self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
- mactobinary(self.pg8.remote_mac),
+ mac_pton(self.pg8.remote_mac),
self.pg8.remote_ip4n,
is_static=1)
self.icmp_id_out = 30608
self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- mactobinary(self.pg7.remote_mac),
+ mac_pton(self.pg7.remote_mac),
self.pg7.remote_ip4n,
is_static=1)
self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
- mactobinary(self.pg8.remote_mac),
+ mac_pton(self.pg8.remote_mac),
self.pg8.remote_ip4n,
is_static=1)
from vpp_sub_interface import VppP2PSubint
from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath
-from vpp_mac import mactobinary
+from vpp_papi import mac_pton
class P2PEthernetAPI(VppTestCase):
i.admin_up()
def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
- p2p = VppP2PSubint(self, parent_if, sub_id, mactobinary(remote_mac))
+ p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
self.p2p_sub_ifs.append(p2p)
def delete_p2p_ethernet(self, parent_if, remote_mac):
self.vapi.delete_p2pethernet_subif(parent_if.sw_if_index,
- mactobinary(remote_mac))
+ mac_pton(remote_mac))
def test_api(self):
"""delete/create p2p subif"""
try:
macs.append(':'.join(re.findall('..', '{:02x}'.format(mac+i))))
self.vapi.create_p2pethernet_subif(self.pg2.sw_if_index,
- mactobinary(macs[i-1]),
+ mac_pton(macs[i-1]),
i)
except Exception:
self.logger.info("Failed to create subif %d %s" % (
super(P2PEthernetIPV6, self).tearDown()
def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
- p2p = VppP2PSubint(self, parent_if, sub_id, mactobinary(remote_mac))
+ p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
p2p.admin_up()
p2p.config_ip6()
p2p.disable_ipv6_ra()
return dst_if.get_capture(count)
def create_p2p_ethernet(self, parent_if, sub_id, remote_mac):
- p2p = VppP2PSubint(self, parent_if, sub_id, mactobinary(remote_mac))
+ p2p = VppP2PSubint(self, parent_if, sub_id, mac_pton(remote_mac))
p2p.admin_up()
p2p.config_ip4()
return p2p
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
from scapy.volatile import RandMAC, RandIP
-from vpp_mac import mactobinary
from util import ppp, ppc
import socket
import unittest
from framework import VppTestCase, VppTestRunner
-from vpp_mac import mactobinary, binarytomac
+from vpp_papi import mac_pton, mac_ntop
class TestUtil (VppTestCase):
""" MAC to binary and back """
def test_mac_to_binary(self):
mac = 'aa:bb:cc:dd:ee:ff'
- b = mactobinary(mac)
- mac2 = binarytomac(b)
+ b = mac_pton(mac)
+ mac2 = mac_ntop(b)
self.assertEqual(type(mac), type(mac2))
self.assertEqual(mac2, mac)
from scapy.utils import hexdump
from socket import AF_INET6
from io import BytesIO
-from vpp_mac import mactobinary
+from vpp_papi import mac_pton
def ppp(headline, packet):
@property
def bin_mac(self):
""" MAC address """
- return mactobinary(self._mac)
+ return mac_pton(self._mac)
@property
def ip4(self):
from six import moves
from util import Host, mk_ll_addr
-from vpp_mac import mactobinary, binarytomac
+from vpp_papi import mac_pton, mac_ntop
class VppInterface(object):
self._hosts_by_ip6 = {}
def set_mac(self, mac):
- self._local_mac = mac
- self._local_ip6_ll = mk_ll_addr(mac)
+ self._local_mac = str(mac)
+ self._local_ip6_ll = mk_ll_addr(self._local_mac)
self.test.vapi.sw_interface_set_mac_address(
- self.sw_if_index,
- mactobinary(self._local_mac))
+ self.sw_if_index, mac.packed)
def set_sw_if_index(self, sw_if_index):
self._sw_if_index = sw_if_index
if intf.sw_if_index == self.sw_if_index:
self._name = intf.interface_name.split(b'\0',
1)[0].decode('utf8')
- self._local_mac = binarytomac(intf.l2_address)
+ self._local_mac = mac_ntop(intf.l2_address)
self._dump = intf
break
else:
:param vrf_id: The FIB table / VRF ID. (Default value = 0)
"""
for host in self._remote_hosts:
- macn = mactobinary(host.mac)
+ macn = mac_pton(host.mac)
ipn = host.ip4n
self.test.vapi.ip_neighbor_add_del(
self.sw_if_index, macn, ipn)
:param vrf_id: The FIB table / VRF ID. (Default value = 0)
"""
for host in self._remote_hosts:
- macn = mactobinary(host.mac)
+ macn = mac_pton(host.mac)
ipn = host.ip6n
self.test.vapi.ip_neighbor_add_del(
self.sw_if_index, macn, ipn, is_ipv6=1)
from vpp_object import *
from vpp_ip import VppIpAddress
-from vpp_mac import VppMacAddress, mactobinary
from vpp_lo_interface import VppLoInterface
+from vpp_papi import MACAddress
class L2_PORT_TYPE:
def find_bridge_domain_arp_entry(test, bd_id, mac, ip):
- vmac = VppMacAddress(mac)
+ vmac = MACAddress(mac)
vip = VppIpAddress(ip)
if vip.version == 4:
arps = test.vapi.bd_ip_mac_dump(bd_id)
for arp in arps:
# do IP addr comparison too once .api is fixed...
- if vmac.bytes == arp.mac_address and \
+ if vmac.packed == arp.mac_address and \
vip.bytes == arp.ip_address[:n]:
return True
return False
def find_l2_fib_entry(test, bd_id, mac, sw_if_index):
- vmac = VppMacAddress(mac)
+ vmac = MACAddress(mac)
lfs = test.vapi.l2_fib_table_dump(bd_id)
for lf in lfs:
- if vmac.bytes == lf.mac and sw_if_index == lf.sw_if_index:
+ if vmac.packed == lf.mac and sw_if_index == lf.sw_if_index:
return True
return False
def __init__(self, test, bd, mac, ip):
self._test = test
self.bd = bd
- self.mac = VppMacAddress(mac)
+ self.mac = MACAddress(mac)
self.ip = VppIpAddress(ip)
def add_vpp_config(self):
self._test.vapi.bd_ip_mac_add_del(
self.bd.bd_id,
- self.mac.encode(),
+ self.mac.packed,
self.ip.encode(),
is_add=1)
self._test.registry.register(self, self._test.logger)
def remove_vpp_config(self):
self._test.vapi.bd_ip_mac_add_del(
self.bd.bd_id,
- self.mac.encode(),
+ self.mac.packed,
self.ip.encode(),
is_add=0)
def query_vpp_config(self):
return find_bridge_domain_arp_entry(self._test,
self.bd.bd_id,
- self.mac.address,
+ self.mac.packed,
self.ip.address)
def __str__(self):
static_mac=0, filter_mac=0, bvi_mac=-1):
self._test = test
self.bd = bd
- self.mac = VppMacAddress(mac)
+ self.mac = MACAddress(mac)
self.itf = itf
self.static_mac = static_mac
self.filter_mac = filter_mac
def add_vpp_config(self):
self._test.vapi.l2fib_add_del(
- self.mac.address,
+ self.mac.packed,
self.bd.bd_id,
self.itf.sw_if_index,
is_add=1,
def remove_vpp_config(self):
self._test.vapi.l2fib_add_del(
- self.mac.address,
+ self.mac.packed,
self.bd.bd_id,
self.itf.sw_if_index,
is_add=0)
def query_vpp_config(self):
return find_l2_fib_entry(self._test,
self.bd.bd_id,
- self.mac.address,
+ self.mac.packed,
self.itf.sw_if_index)
def __str__(self):
+++ /dev/null
-"""
- MAC Types
-
-"""
-import binascii
-
-
-def mactobinary(mac):
- """ Convert the : separated format into binary packet data for the API """
- return binascii.unhexlify(mac.replace(':', ''))
-
-
-def binarytomac(binary):
- """ Convert binary packed data in a : separated string """
- x = b':'.join(binascii.hexlify(binary)[i:i + 2]
- for i in range(0, 12, 2))
- return str(x.decode('ascii'))
-
-
-class VppMacAddress():
- def __init__(self, addr):
- self._address = addr
-
- def encode(self):
- return {
- 'bytes': self.bytes
- }
-
- @property
- def bytes(self):
- return mactobinary(self.address)
-
- @property
- def address(self):
- return self._address
-
- def __str__(self):
- return self.address
-
- def __eq__(self, other):
- if isinstance(other, self.__class__):
- return self.address == other.address
- elif hasattr(other, "bytes"):
- # vl_api_mac_addres_t
- return self.bytes == other.bytes
- else:
- raise TypeError("Comparing VppMacAddress:%s"
- "with unknown type: %s" %
- (self, other))
- return False
from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
from vpp_object import *
-from vpp_mac import mactobinary
+from vpp_papi import mac_pton
def find_nbr(test, sw_if_index, ip_addr, is_static=0, inet=AF_INET, mac=None):
if nbr_addr == n.ip_address[:s] \
and is_static == n.is_static:
if mac:
- if n.mac_address == mactobinary(mac):
+ if n.mac_address == mac_pton(mac):
return True
else:
return True
af=AF_INET, is_static=False, is_no_fib_entry=0):
self._test = test
self.sw_if_index = sw_if_index
- self.mac_addr = mactobinary(mac_addr)
+ self.mac_addr = mac_pton(mac_addr)
self.af = af
self.is_static = is_static
self.is_no_fib_entry = is_no_fib_entry
from collections import deque
from six import moves
-from vpp_mac import mactobinary
+from vpp_papi import mac_pton
from hook import Hook
from vpp_l2 import L2_PORT_TYPE
"""
return cli + "\n" + str(self.cli(cli))
- def _convert_mac(self, mac):
- return mactobinary(mac)
-
def show_version(self):
""" """
return self.api(self.papi.show_version, {})
interface. (Default value = 0)
"""
return self.api(self.papi.l2fib_add_del,
- {'mac': self._convert_mac(mac),
+ {'mac': mac,
'bd_id': bd_id,
'sw_if_index': sw_if_index,
'is_add': is_add,
from vpp_interface import VppInterface
import socket
-from vpp_mac import mactobinary
+from vpp_papi import mac_pton
class VppPppoeInterface(VppInterface):
def add_vpp_config(self):
cip = socket.inet_pton(socket.AF_INET, self.client_ip)
- cmac = mactobinary(self.client_mac)
+ cmac = mac_pton(self.client_mac)
r = self.test.vapi.pppoe_add_del_session(
cip, cmac,
session_id=self.session_id,
def remove_vpp_config(self):
cip = socket.inet_pton(socket.AF_INET, self.client_ip)
- cmac = mactobinary(self.client_mac)
+ cmac = mac_pton(self.client_mac)
self.unconfig()
self.test.vapi.pppoe_add_del_session(
cip, cmac,