#!/usr/bin/env python import unittest import socket import struct from framework import VppTestCase, VppTestRunner, running_extended_tests from vpp_neighbor import VppNeighbor from vpp_ip_route import find_route, VppIpTable from util import mk_ll_addr 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 scapy.utils6 import in6_mactoifaceid from scapy.layers.dhcp import DHCP, BOOTP, DHCPTypes from scapy.layers.dhcp6 import DHCP6, DHCP6_Solicit, DHCP6_RelayForward, \ DHCP6_RelayReply, DHCP6_Advertise, DHCP6OptRelayMsg, DHCP6OptIfaceId, \ DHCP6OptStatusCode, DHCP6OptVSS, DHCP6OptClientLinkLayerAddr, DHCP6_Request from socket import AF_INET, AF_INET6 from scapy.utils import inet_pton, inet_ntop from scapy.utils6 import in6_ptop from util import mactobinary DHCP4_CLIENT_PORT = 68 DHCP4_SERVER_PORT = 67 DHCP6_CLIENT_PORT = 547 DHCP6_SERVER_PORT = 546 class TestDHCP(VppTestCase): """ DHCP Test Case """ def setUp(self): super(TestDHCP, self).setUp() # create 6 pg interfaces for pg0 to pg5 self.create_pg_interfaces(range(6)) self.tables = [] # pg0 to 2 are IP configured in VRF 0, 1 and 2. # pg3 to 5 are non IP-configured in VRF 0, 1 and 2. table_id = 0 for table_id in range(1, 4): tbl4 = VppIpTable(self, table_id) tbl4.add_vpp_config() self.tables.append(tbl4) tbl6 = VppIpTable(self, table_id, is_ip6=1) tbl6.add_vpp_config() self.tables.append(tbl6) table_id = 0 for i in self.pg_interfaces[:3]: i.admin_up() i.set_table_ip4(table_id) i.set_table_ip6(table_id) i.config_ip4() i.resolve_arp() i.config_ip6() i.resolve_ndp() table_id += 1 table_id = 0 for i in self.pg_interfaces[3:]: i.admin_up() i.set_table_ip4(table_id) i.set_table_ip6(table_id) table_id += 1 def tearDown(self): for i in self.pg_interfaces[:3]: i.unconfig_ip4() i.unconfig_ip6() for i in self.pg_interfaces: i.set_table_ip4(0) i.set_table_ip6(0) i.admin_down() super(TestDHCP, self).tearDown() def verify_dhcp_has_option(self, pkt, option, value): dhcp = pkt[DHCP] found = False for i in dhcp.options: if type(i) is tuple: if i[0] == option: self.assertEqual(i[1], value) found = True self.assertTrue(found) def validate_relay_options(self, pkt, intf, ip_addr, vpn_id, fib_id, oui): dhcp = pkt[DHCP] found = 0 data = [] id_len = len(vpn_id) for i in dhcp.options: if type(i) is tuple: if i[0] == "relay_agent_Information": # # There are two sb-options present - each of length 6. # data = i[1] if oui != 0: self.assertEqual(len(data), 24) elif len(vpn_id) > 0: self.assertEqual(len(data), len(vpn_id)+17) else: self.assertEqual(len(data), 12) # # First sub-option is ID 1, len 4, then encoded # sw_if_index. This test uses low valued indicies # so [2:4] are 0. # The ID space is VPP internal - so no matching value # scapy # self.assertEqual(ord(data[0]), 1) self.assertEqual(ord(data[1]), 4) self.assertEqual(ord(data[2]), 0) self.assertEqual(ord(data[3]), 0) self.assertEqual(ord(data[4]), 0) self.assertEqual(ord(data[5]), intf._sw_if_index) # # next sub-option is the IP address of the client side # interface. # sub-option ID=5, length (of a v4 address)=4 # claddr = socket.inet_pton(AF_INET, ip_addr) self.assertEqual(ord(data[6]), 5) self.assertEqual(ord(data[7]), 4) self.assertEqual(data[8], claddr[0]) self.assertEqual(data[9], claddr[1]) self.assertEqual(data[10], claddr[2]) self.assertEqual(data[11], claddr[3]) if oui != 0: # sub-option 151 encodes vss_type 1, # the 3 byte oui and the 4 byte fib_id self.assertEqual(id_len, 0) self.assertEqual(ord(data[12]), 151) self.assertEqual(ord(data[13]), 8) self.assertEqual(ord(data[14]), 1) self.assertEqual(ord(data[15]), 0) self.assertEqual(ord(data[16]), 0) self.assertEqual(ord(data[17]), oui) self.assertEqual(ord(data[18]), 0) self.assertEqual(ord(data[19]), 0) self.assertEqual(ord(data[20]), 0) self.assertEqual(ord(data[21]), fib_id) # VSS control sub-option self.assertEqual(ord(data[22]), 152) self.assertEqual(ord(data[23]), 0) if id_len > 0: # sub-option 151 encode vss_type of 0 # followerd by vpn_id in ascii self.assertEqual(oui, 0) self.assertEqual(ord(data[12]), 151) self.assertEqual(ord(data[13]), id_len+1) self.assertEqual(ord(data[14]), 0) self.assertEqual(data[15:15+id_len], vpn_id) # VSS control sub-option self.assertEqual(ord(data[15+len(vpn_id)]), 152) self.assertEqual(ord(data[16+len(vpn_id)]), 0) found = 1 self.assertTrue(found) return data def verify_dhcp_msg_type(self, pkt, name): dhcp = pkt[DHCP] found = False for o in dhcp.options: if type(o) is tuple: if o[0] == "message-type" \ and DHCPTypes[o[1]] == name: found = True self.assertTrue(found) def verify_dhcp_offer(self, pkt, intf, vpn_id="", fib_id=0, oui=0): ether = pkt[Ether] self.assertEqual(ether.dst, "ff:ff:ff:ff:ff:ff") self.assertEqual(ether.src, intf.local_mac) ip = pkt[IP] self.assertEqual(ip.dst, "255.255.255.255") self.assertEqual(ip.src, intf.local_ip4) udp = pkt[UDP] self.assertEqual(udp.dport, DHCP4_CLIENT_PORT) self.assertEqual(udp.sport, DHCP4_SERVER_PORT) self.verify_dhcp_msg_type(pkt, "offer") data = self.validate_relay_options(pkt, intf, intf.local_ip4, vpn_id, fib_id, oui) def verify_orig_dhcp_pkt(self, pkt, intf): ether = pkt[Ether] self.assertEqual(ether.dst, "ff:ff:ff:ff:ff:ff") self.assertEqual(ether.src, intf.local_mac) ip = pkt[IP] self.assertEqual(ip.dst, "255.255.255.255") self.assertEqual(ip.src, "0.0.0.0") udp = pkt[UDP] self.assertEqual(udp.dport, DHCP4_SERVER_PORT) self.assertEqual(udp.sport, DHCP4_CLIENT_PORT) def verify_orig_dhcp_discover(self, pkt, intf, hostname, client_id=None, broadcast=1): self.verify_orig_dhcp_pkt(pkt, intf) self.verify_dhcp_msg_type(pkt, "discover") self.verify_dhcp_has_option(pkt, "hostname", hostname) if client_id: self.verify_dhcp_has_option(pkt, "client_id", client_id) bootp = pkt[BOOTP] self.assertEqual(bootp.ciaddr, "0.0.0.0") self.assertEqual(bootp.giaddr, "0.0.0.0") if broadcast: self.assertEqual(bootp.flags, 0x8000) else: self.assertEqual(bootp.flags, 0x0000) def verify_orig_dhcp_request(self, pkt, intf, hostname, ip, broadcast=1):
/* * Copyright (c) 2011-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.
*/
/**
* @file
* @brief BFD UDP transport layer declarations
*/
#ifndef __included_bfd_udp_h__
#define __included_bfd_udp_h__
#include <vppinfra/clib.h>
#include <vnet/adj/adj_types.h>
#include <vnet/ip/ip6_packet.h>
#include <vnet/bfd/bfd_api.h>
/* *INDENT-OFF* */
/** identifier of BFD session based on UDP transport only */
typedef CLIB_PACKED (struct {
union {
/** interface to which the session is tied - single-hop */
u32 sw_if_index;
/** the FIB index the peer is in - multi-hop*/
u32 fib_index;
};
/** local address */
ip46_address_t local_addr;
/** peer address */
ip46_address_t peer_addr;
}) bfd_udp_key_t;
/* *INDENT-ON* */
/** UDP transport specific data embedded in bfd_session's union */
typedef struct
{
/** key identifying this session */
bfd_udp_key_t key;
/** adjacency index returned from adj lock call */
adj_index_t adj_index;
} bfd_udp_session_t;
/** bfd udp echo packet trace capture */
typedef struct
{
u32 len;
u8 data[400];
} bfd_udp_echo_input_trace_t;
struct bfd_session_s;
/**
* @brief add the necessary transport layer by prepending it to existing data
*
*
* @param is_echo 1 if this is echo packet, 0 if control frame
*
* @return 1 on success, 0 on failure
*/
int bfd_add_udp4_transport (vlib_main_t * vm, u32 bi,
const struct bfd_session_s *bs, int is_echo);
/**
* @brief add the necessary transport layer by prepending it to existing data
*
* @param is_echo 1 if this is echo packet, 0 if control frame
*
* @return 1 on success, 0 on failure
*/
int bfd_add_udp6_transport (vlib_main_t * vm, u32 bi,
const struct bfd_session_s *bs, int is_echo);
/**
* @brief transport packet over udpv4
*
* @param is_echo 1 if this is echo packet, 0 if control frame
*
* @return 1 on success, 0 on failure
*/
int bfd_transport_udp4 (vlib_main_t * vm, u32 bi,
const struct bfd_session_s *bs);
/**
* @brief transport packet over udpv6
*
* @param is_echo 1 if this is echo packet, 0 if control frame
*
* @return 1 on success, 0 on failure
*/
int bfd_transport_udp6 (vlib_main_t * vm, u32 bi,
const struct bfd_session_s *bs);
/**
* @brief check if the bfd udp layer is echo-capable at this time
*
* @return 1 if available, 0 otherwise
*/
int bfd_udp_is_echo_available (bfd_transport_e transport);
/**
* @brief get echo source information - used by CLI
*/
void bfd_udp_get_echo_source (int *is_set, u32 * sw_if_index,
int *have_usable_ip4, ip4_address_t * ip4,
int *have_usable_ip6, ip6_address_t * ip6);
#endif /* __included_bfd_udp_h__ */
/*
* fd.io coding-style-patch-verification: ON
*
* Local Variables:
* eval: (c-set-style "gnu")
* End:
*/