# NB NB NB NB NB NB NB NB NB NB NB # # NOTE: The API binary wrappers in this file are in the process of being # deprecated. DO NOT ADD NEW WRAPPERS HERE. Call the functions using # named arguments directly instead. # import os import socket import time import queue from six import moves, iteritems from config import config from vpp_papi import VPPApiClient from hook import Hook from vpp_papi_exceptions import ( CliFailedCommandError, CliSyntaxError, UnexpectedApiReturnValueError, ) # # Dictionary keyed on message name to override default values for # named parameters # defaultmapping = { "acl_interface_add_del": {"is_add": 1, "is_input": 1}, "bd_ip_mac_add_del": { "is_add": 1, }, "bfd_udp_add": {"is_authenticated": False, "bfd_key_id": None, "conf_key_id": None}, "bfd_udp_auth_activate": { "bfd_key_id": None, "conf_key_id": None, "is_delayed": False, }, "bier_disp_entry_add_del": { "next_hop_rpf_id": -1, "next_hop_is_ip4": 1, "is_add": 1, }, "bier_disp_table_add_del": { "is_add": 1, }, "bier_imp_add": { "is_add": 1, }, "bier_route_add_del": { "is_add": 1, }, "bier_table_add_del": { "is_add": 1, }, "bvi_delete": {}, "geneve_add_del_tunnel": { "mcast_sw_if_index": 4294967295, "is_add": 1, "decap_next_index": 4294967295, }, "input_acl_set_interface": { "ip4_table_index": 4294967295, "ip6_table_index": 4294967295, "l2_table_index": 4294967295, }, "ip6_add_del_address_using_prefix": { "is_add": 1, }, "ip6nd_send_router_solicitation": { "irt": 1, "mrt": 120, }, "ip_add_del_route": { "next_hop_sw_if_index": 4294967295, "next_hop_weight": 1, "next_hop_via_label": 1048576, "classify_table_index": 4294967295, "is_add": 1, }, "ip_mroute_add_del": { "is_add": 1, }, "ip_neighbor_add_del": { "is_add": 1, }, "ipsec_interface_add_del_spd": { "is_add": 1, }, "ipsec_spd_add_del": { "is_add": 1, }, "ipsec_spd_dump": { "sa_id": 4294967295, }, "ipsec_spd_entry_add_del": { "local_port_stop": 65535, "remote_port_stop": 65535, "priority": 100, "is_outbound": 1, "is_add": 1, }, "ipsec_tunnel_if_add_del": { "is_add": 1, "anti_replay": 1, }, "l2_emulation": { "enable": 1, }, "l2fib_add_del": { "is_add": 1, }, "lisp_add_del_adjacency": { "is_add": 1, }, "lisp_add_del_local_eid": { "is_add": 1, }, "lisp_add_del_locator": { "priority": 1, "weight": 1, "is_add": 1, }, "lisp_add_del_locator_set": { "is_add": 1, }, "lisp_add_del_remote_mapping": { "is_add": 1, }, "macip_acl_interface_add_del": { "is_add": 1, }, "mpls_ip_bind_unbind": { "is_ip4": 1, "is_bind": 1, }, "mpls_route_add_del": { "mr_next_hop_sw_if_index": 4294967295, "mr_next_hop_weight": 1, "mr_next_hop_via_label": 1048576, "mr_is_add": 1, "mr_classify_table_index": 4294967295, }, "mpls_table_add_del": { "is_add": 1, }, "mpls_tunnel_add_del": { "next_hop_sw_if_index": 4294967295, "next_hop_weight": 1, "next_hop_via_label": 1048576, "is_add": 1, }, "output_acl_set_interface": { "ip4_table_index": 4294967295, "ip6_table_index": 4294967295, "l2_table_index": 4294967295, }, "pppoe_add_del_session": { "is_add": 1, }, "policer_add_del": { "is_add": 1, "conform_action": {"type": 1}, }, "set_ipfix_exporter": { "collector_port": 4739, }, "sr_policy_add": { "weight": 1, "is_encap": 1, }, "sw_interface_add_del_address": { "is_add": 1, }, "sw_interface_ip6nd_ra_prefix": { "val_lifetime": 4294967295, "pref_lifetime": 4294967295, }, "sw_interface_set_ip_directed_broadcast": { "enable": 1, }, "sw_interface_set_l2_bridge": { "enable": 1, }, "sw_interface_set_mpls_enable": { "enable": 1, }, "sw_interface_set_mtu": { "mtu": [0, 0, 0, 0], }, "sw_interface_set_unnumbered": { "is_add": 1, }, "sw_interface_span_enable_disable": { "state": 1, }, "vxlan_add_del_tunnel": { "mcast_sw_if_index": 4294967295, "is_add": 1, "decap_next_index": 4294967295, "instance": 4294967295, }, "want_bfd_events": { "enable_disable": 1, }, "want_igmp_events": { "enable": 1, }, "want_interface_events": { "enable_disable": 1, }, "want_l2_macs_events": { "enable_disable": 1, "pid": os.getpid(), }, "want_l2_macs_events2": { "enable_disable": 1, "pid": os.getpid(), }, } def as_fn_signature(d): return ", ".join(f"{k}={v}" for k, v in d.items()) class VppPapiProvider(object): """VPP-api provider using vpp-papi @property hook: hook object providing before and after api/cli hooks """ _zero, _negative = range(2) def __init__(self, name, test_class, read_timeout): self.hook = Hook(test_class) self.name = name self.test_class = test_class self._expect_api_retval = self._zero self._expect_stack = [] self.vpp = VPPApiClient( apidir=config.extern_apidir + [config.vpp_install_dir], logger=test_class.logger, read_timeout=read_timeout, use_socket=True, server_address=test_class.get_api_sock_path(), ) self._events = queue.Queue() def __enter__(self): return self def assert_negative_api_retval(self): """Expect API failure - used with with, e.g.:: with self.vapi.assert_negative_api_retval(): self.vapi. .. """ self._expect_stack.append(self._expect_api_retval) self._expect_api_retval = self._negative return self def assert_zero_api_retval(self): """Expect API success - used with with, e.g.:: with self.vapi.assert_negative_api_retval(): self.vapi. :note: this is useful only inside another with block as success is the default expected value """ self._expect_stack.append(self._expect_api_retval) self._expect_api_retval = self._zero return self def __exit__(self, exc_type, exc_value, traceback): self._expect_api_retval = self._expect_stack.pop() def register_hook(self, hook): """Replace hook registration with new hook :param hook: """ self.hook = hook def collect_events(self): """Collect all events from the internal queue and clear the queue.""" result = [] while True: try: e = self._events.get(block=False) result.append(e) except queue.Empty: return result return result def wait_for_event(self, timeout, name=None): """Wait for and return next event.""" if name: self.test_class.logger.debug( "Expecting event '%s' within %ss", name, timeout ) else: self.test_class.logger.debug("Expecting event within %ss", timeout) try: e = self._events.get(timeout=timeout) except queue.Empty: raise Exception("Event did not occur within timeout") msgname = type(e).__name__ if name and msgname != name: raise Exception("Unexpected event received: %s, expected: %s" % msgname) self.test_class.logger.debug("Returning event %s:%s" % (name, e)) return e def __call__(self, name, event): """Enqueue event in the internal event queue.""" self.test_class.logger.debug("New event: %s: %s" % (name, event)) self._events.put(event) def factory(self, name, apifn): def f(*a, **ka): fields = apifn._func.msg.fields # add positional and kw arguments d = ka for i, o in enumerate(fields[3:]): try: d[o] = a[i] except BaseException: 
/*
 * Copyright (c) 2015-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.
 */

/** \brief l2tpv3 tunnel interface create request
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param client_address - remote client tunnel ip address
    @param client_address - local tunnel ip address
    @param is_ipv6 - ipv6 if non-zero, else ipv4
    @param local_session_id - local tunnel session id
    @param remote_session_id - remote tunnel session id
    @param local_cookie - local tunnel cookie
    @param l2_sublayer_present - l2 sublayer is present in packets if non-zero
    @param encap_vrf_id - fib identifier used for outgoing encapsulated packets
*/
define l2tpv3_create_tunnel
{
  u32 client_index;
  u32 context;
  u8 client_address[16];
  u8 our_address[16];
  u8 is_ipv6;
  u32 local_session_id;
  u32 remote_session_id;
  u64 local_cookie;
  u64 remote_cookie;
  u8 l2_sublayer_present;
  u32 encap_vrf_id;
};

/** \brief l2tpv3 tunnel interface create response
    @param context - sender context, to match reply w/ request
    @param retval - return code for the request
    @param sw_if_index - index of the new tunnel interface
*/
define l2tpv3_create_tunnel_reply
{
  u32 context;
  i32 retval;
  u32 sw_if_index;
};

define l2tpv3_set_tunnel_cookies
{
  u32 client_index;
  u32 context;
  u32 sw_if_index;
  u64 new_local_cookie;
  u64 new_remote_cookie;
};

/** \brief L2TP tunnel set cookies response
    @param context - sender context, to match reply w/ request
    @param retval - return code for the request
*/
define l2tpv3_set_tunnel_cookies_reply
{
  u32 context;
  i32 retval;
};

define sw_if_l2tpv3_tunnel_details
{
  u32 context;
  u32 sw_if_index;
  u8 interface_name[64];
  u8 client_address[16];
  u8 our_address[16];
  u32 local_session_id;
  u32 remote_session_id;
  u64 local_cookie[2];
  u64 remote_cookie;
  u8 l2_sublayer_present;
};

define sw_if_l2tpv3_tunnel_dump
{
  u32 client_index;
  u32 context;
};

define l2tpv3_interface_enable_disable
{
  u32 client_index;
  u32 context;
  u8 enable_disable;
  u32 sw_if_index;
};

define l2tpv3_interface_enable_disable_reply
{
  u32 context;
  i32 retval;
};

define l2tpv3_set_lookup_key
{
  u32 client_index;
  u32 context;
  /* 0 = ip6 src_address, 1 = ip6 dst_address, 2 = session_id */
  u8 key;
};

define l2tpv3_set_lookup_key_reply
{
  u32 context;
  i32 retval;
};

/*
 * Local Variables:
 * eval: (c-set-style "gnu")
 * End:
 */
id, "path_mtu": path_mtu, "template_interval": template_interval, "udp_checksum": udp_checksum, }, ) def mfib_signal_dump(self): return self.api(self.papi.mfib_signal_dump, {}) def ip_mroute_dump(self, table_id, is_ip6=False): return self.api( self.papi.ip_mroute_dump, {"table": {"table_id": table_id, "is_ip6": is_ip6}}, ) def pppoe_add_del_session( self, client_ip, client_mac, session_id=0, is_add=1, decap_vrf_id=0 ): """ :param is_add: (Default value = 1) :param is_ipv6: (Default value = 0) :param client_ip: :param session_id: (Default value = 0) :param client_mac: :param decap_vrf_id: (Default value = 0) """ return self.api( self.papi.pppoe_add_del_session, { "is_add": is_add, "session_id": session_id, "client_ip": client_ip, "decap_vrf_id": decap_vrf_id, "client_mac": client_mac, }, ) def sr_mpls_policy_add(self, bsid, weight, type, segments): return self.api( self.papi.sr_mpls_policy_add, { "bsid": bsid, "weight": weight, "is_spray": type, "n_segments": len(segments), "segments": segments, }, ) def sr_mpls_policy_del(self, bsid): return self.api(self.papi.sr_mpls_policy_del, {"bsid": bsid}) def bier_table_add_del(self, bti, mpls_label, is_add=1): """BIER Table add/del""" return self.api( self.papi.bier_table_add_del, { "bt_tbl_id": { "bt_set": bti.set_id, "bt_sub_domain": bti.sub_domain_id, "bt_hdr_len_id": bti.hdr_len_id, }, "bt_label": mpls_label, "bt_is_add": is_add, }, ) def bier_table_dump(self): return self.api(self.papi.bier_table_dump, {}) def bier_route_add_del(self, bti, bp, paths, is_add=1, is_replace=0): """BIER Route add/del""" return self.api( self.papi.bier_route_add_del, { "br_route": { "br_tbl_id": { "bt_set": bti.set_id, "bt_sub_domain": bti.sub_domain_id, "bt_hdr_len_id": bti.hdr_len_id, }, "br_bp": bp, "br_n_paths": len(paths), "br_paths": paths, }, "br_is_add": is_add, "br_is_replace": is_replace, }, ) def bier_route_dump(self, bti): return self.api( self.papi.bier_route_dump, { "br_tbl_id": { "bt_set": bti.set_id, "bt_sub_domain": bti.sub_domain_id, "bt_hdr_len_id": bti.hdr_len_id, } }, ) def bier_imp_add(self, bti, src, ibytes, is_add=1): """BIER Imposition Add""" return self.api( self.papi.bier_imp_add, { "bi_tbl_id": { "bt_set": bti.set_id, "bt_sub_domain": bti.sub_domain_id, "bt_hdr_len_id": bti.hdr_len_id, }, "bi_src": src, "bi_n_bytes": len(ibytes), "bi_bytes": ibytes, }, ) def bier_imp_del(self, bi_index): """BIER Imposition del""" return self.api(self.papi.bier_imp_del, {"bi_index": bi_index}) def bier_imp_dump(self): return self.api(self.papi.bier_imp_dump, {}) def bier_disp_table_add_del(self, bdti, is_add=1): """BIER Disposition Table add/del""" return self.api( self.papi.bier_disp_table_add_del, {"bdt_tbl_id": bdti, "bdt_is_add": is_add}, ) def bier_disp_table_dump(self): return self.api(self.papi.bier_disp_table_dump, {}) def bier_disp_entry_add_del( self, bdti, bp, payload_proto, next_hop_afi, next_hop, next_hop_tbl_id=0, next_hop_rpf_id=~0, next_hop_is_ip4=1, is_add=1, ): """BIER Route add/del""" lstack = [] while len(lstack) < 16: lstack.append({}) return self.api( self.papi.bier_disp_entry_add_del, { "bde_tbl_id": bdti, "bde_bp": bp, "bde_payload_proto": payload_proto, "bde_n_paths": 1, "bde_paths": [ { "table_id": next_hop_tbl_id, "rpf_id": next_hop_rpf_id, "n_labels": 0, "label_stack": lstack, } ], "bde_is_add": is_add, }, ) def bier_disp_entry_dump(self, bdti): return self.api(self.papi.bier_disp_entry_dump, {"bde_tbl_id": bdti}) def ipsec_spd_add_del(self, spd_id, is_add=1): """SPD add/del - Wrapper to add or del ipsec SPD Sample CLI : 'ipsec spd add 1' :param spd_id - SPD ID to be created in the vpp . mandatory :param is_add - create (1) or delete(0) SPD (Default 1 - add) . optional :returns: reply from the API """ return self.api( self.papi.ipsec_spd_add_del, {"spd_id": spd_id, "is_add": is_add} ) def ipsec_spds_dump(self): return self.api(self.papi.ipsec_spds_dump, {}) def ipsec_interface_add_del_spd(self, spd_id, sw_if_index, is_add=1): """ IPSEC interface SPD add/del - \ Wrapper to associate/disassociate SPD to interface in VPP Sample CLI : 'set interface ipsec spd GigabitEthernet0/6/0 1' :param spd_id - SPD ID to associate with the interface . mandatory :param sw_if_index - Interface Index which needs to ipsec \ association mandatory :param is_add - add(1) or del(0) association with interface \ (Default 1 - add) . optional :returns: reply from the API """ return self.api( self.papi.ipsec_interface_add_del_spd, {"spd_id": spd_id, "sw_if_index": sw_if_index, "is_add": is_add}, ) def ipsec_spd_interface_dump(self, spd_index=None): return self.api( self.papi.ipsec_spd_interface_dump, { "spd_index": spd_index if spd_index else 0, "spd_index_valid": 1 if spd_index else 0, }, ) def ipsec_spd_entry_add_del( self, spd_id, sa_id, local_address_start, local_address_stop, remote_address_start, remote_address_stop, local_port_start=0, local_port_stop=65535, remote_port_start=0, remote_port_stop=65535, protocol=socket.IPPROTO_RAW, policy=0, priority=100, is_outbound=1, is_add=1, is_ipv6=0, is_ip_any=0, ): """IPSEC policy SPD add/del - Wrapper to configure ipsec SPD policy entries in VPP :param spd_id: SPD ID for the policy :param local_address_start: local-ip-range start address :param local_address_stop: local-ip-range stop address :param remote_address_start: remote-ip-range start address :param remote_address_stop: remote-ip-range stop address :param local_port_start: (Default value = 0) :param local_port_stop: (Default value = 65535) :param remote_port_start: (Default value = 0) :param remote_port_stop: (Default value = 65535) :param protocol: Any(0), AH(51) & ESP(50) protocol (Default value = 0) :param sa_id: Security Association ID for mapping it to SPD :param policy: bypass(0), discard(1), resolve(2) or protect(3) action (Default value = 0) :param priority: value for the spd action (Default value = 100) :param is_outbound: flag for inbound(0) or outbound(1) (Default value = 1) :param is_add: (Default value = 1) """ return self.api( self.papi.ipsec_spd_entry_add_del_v2, { "is_add": is_add, "entry": { "spd_id": spd_id, "sa_id": sa_id, "local_address_start": local_address_start, "local_address_stop": local_address_stop, "remote_address_start": remote_address_start, "remote_address_stop": remote_address_stop, "local_port_start": local_port_start, "local_port_stop": local_port_stop, "remote_port_start": remote_port_start, "remote_port_stop": remote_port_stop, "protocol": protocol, "policy": policy, "priority": priority, "is_outbound": is_outbound, }, }, ) def ipsec_spd_dump(self, spd_id, sa_id=0xFFFFFFFF): return self.api(self.papi.ipsec_spd_dump, {"spd_id": spd_id, "sa_id": sa_id}) def ipsec_tunnel_if_add_del( self, local_ip, remote_ip, local_spi, remote_spi, crypto_alg, local_crypto_key, remote_crypto_key, integ_alg, local_integ_key, remote_integ_key, is_add=1, esn=0, salt=0, anti_replay=1, renumber=0, udp_encap=0, show_instance=0xFFFFFFFF, ): return self.api( self.papi.ipsec_tunnel_if_add_del, { "local_ip": local_ip, "remote_ip": remote_ip, "local_spi": local_spi, "remote_spi": remote_spi, "crypto_alg": crypto_alg, "local_crypto_key_len": len(local_crypto_key), "local_crypto_key": local_crypto_key, "remote_crypto_key_len": len(remote_crypto_key), "remote_crypto_key": remote_crypto_key, "integ_alg": integ_alg, "local_integ_key_len": len(local_integ_key), "local_integ_key": local_integ_key, "remote_integ_key_len": len(remote_integ_key), "remote_integ_key": remote_integ_key, "is_add": is_add, "esn": esn, "anti_replay": anti_replay, "renumber": renumber, "show_instance": show_instance, "udp_encap": udp_encap, "salt": salt, }, ) def ipsec_select_backend(self, protocol, index): return self.api( self.papi.ipsec_select_backend, {"protocol": protocol, "index": index} ) def ipsec_backend_dump(self): return self.api(self.papi.ipsec_backend_dump, {}) def punt_socket_register(self, reg, pathname, header_version=1): """Register punt socket""" return self.api( self.papi.punt_socket_register, {"header_version": header_version, "punt": reg, "pathname": pathname}, ) def punt_socket_deregister(self, reg): """Unregister punt socket""" return self.api(self.papi.punt_socket_deregister, {"punt": reg}) def igmp_enable_disable(self, sw_if_index, enable, host): """Enable/disable IGMP on a given interface""" return self.api( self.papi.igmp_enable_disable, {"enable": enable, "mode": host, "sw_if_index": sw_if_index}, ) def igmp_proxy_device_add_del(self, vrf_id, sw_if_index, add): """Add/del IGMP proxy device""" return self.api( self.papi.igmp_proxy_device_add_del, {"vrf_id": vrf_id, "sw_if_index": sw_if_index, "add": add}, ) def igmp_proxy_device_add_del_interface(self, vrf_id, sw_if_index, add): """Add/del interface to/from IGMP proxy device""" return self.api( self.papi.igmp_proxy_device_add_del_interface, {"vrf_id": vrf_id, "sw_if_index": sw_if_index, "add": add}, ) def igmp_listen(self, filter, sw_if_index, saddrs, gaddr): """Listen for new (S,G) on specified interface :param enable: add/delas :param sw_if_index: interface sw index :param saddr: source ip4 addr :param gaddr: group ip4 addr """ return self.api( self.papi.igmp_listen, { "group": { "filter": filter, "sw_if_index": sw_if_index, "n_srcs": len(saddrs), "saddrs": saddrs, "gaddr": gaddr, } }, ) def igmp_clear_interface(self, sw_if_index): """Remove all (S,G)s from specified interface doesn't send IGMP report! """ return self.api(self.papi.igmp_clear_interface, {"sw_if_index": sw_if_index}) def want_igmp_events(self, enable=1): return self.api( self.papi.want_igmp_events, {"enable": enable, "pid": os.getpid()} )