diff options
Diffstat (limited to 'resources/traffic_scripts/dhcp/send_dhcp_v6_messages.py')
| -rwxr-xr-x | resources/traffic_scripts/dhcp/send_dhcp_v6_messages.py | 372 |
1 files changed, 0 insertions, 372 deletions
diff --git a/resources/traffic_scripts/dhcp/send_dhcp_v6_messages.py b/resources/traffic_scripts/dhcp/send_dhcp_v6_messages.py deleted file mode 100755 index 09742565ee..0000000000 --- a/resources/traffic_scripts/dhcp/send_dhcp_v6_messages.py +++ /dev/null @@ -1,372 +0,0 @@ -#!/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. - -"""Traffic script that sends DHCPv6 proxy packets.""" - -from scapy.layers.dhcp6 import * -from scapy.layers.inet6 import IPv6, UDP, UDP_SERVICES - -from resources.libraries.python.PacketVerifier import RxQueue, TxQueue -from resources.libraries.python.TrafficScriptArg import TrafficScriptArg - - -def _check_udp_checksum(pkt): - """Check udp checksum in ip packet. - Return true if checksum is correct.""" - new = pkt.__class__(str(pkt)) - del new['UDP'].chksum - new = new.__class__(str(new)) - return new['UDP'].chksum == pkt['UDP'].chksum - - -def _get_dhcpv6_msgtype(msg_index): - """Return DHCPv6 message type string. - - :param msg_index: Index of message type. - :return: Message type. - :type msg_index: int - :rtype msg_str: str - """ - dhcp6_messages = { - 1: "SOLICIT", - 2: "ADVERTISE", - 3: "REQUEST", - 4: "CONFIRM", - 5: "RENEW", - 6: "REBIND", - 7: "REPLY", - 8: "RELEASE", - 9: "DECLINE", - 10: "RECONFIGURE", - 11: "INFORMATION-REQUEST", - 12: "RELAY-FORW", - 13: "RELAY-REPL" - } - return dhcp6_messages[msg_index] - - -def dhcpv6_solicit(tx_if, rx_if, dhcp_multicast_ip, link_local_ip, proxy_ip, - server_ip, server_mac, client_duid, client_mac): - """Send and check DHCPv6 SOLICIT proxy packet. - - :param tx_if: Client interface. - :param rx_if: DHCPv6 server interface. - :param dhcp_multicast_ip: Servers and relay agents multicast address. - :param link_local_ip: Client link-local address. - :param proxy_ip: IP address of DHCPv6 proxy server. - :param server_ip: IP address of DHCPv6 server. - :param server_mac: MAC address of DHCPv6 server. - :param client_duid: Client DHCP Unique Identifier. - :param client_mac: Client MAC address. - :type tx_if: str - :type rx_if: str - :type dhcp_multicast_ip: str - :type link_local_ip: str - :type proxy_ip: str - :type server_ip: str - :type server_mac: str - :type client_duid: str - :type client_mac: str - :return interface_id: ID of proxy interface. - :rtype interface_id: str - """ - - rxq = RxQueue(rx_if) - txq = TxQueue(tx_if) - - sent_packets = [] - - dhcp6_solicit_pkt = Ether(src=client_mac, dst="33:33:00:01:00:02") / \ - IPv6(src=link_local_ip, dst=dhcp_multicast_ip) / \ - UDP(sport=UDP_SERVICES.dhcpv6_client, - dport=UDP_SERVICES.dhcpv6_server) / \ - DHCP6_Solicit() / \ - DHCP6OptClientId(duid=client_duid) - - sent_packets.append(dhcp6_solicit_pkt) - txq.send(dhcp6_solicit_pkt) - - ether = rxq.recv(2) - - if ether is None: - raise RuntimeError('DHCPv6 SOLICIT timeout') - - if ether.dst != server_mac: - raise RuntimeError("Destination MAC address error: {} != {}".format( - ether.dst, server_mac)) - print "Destination MAC address: OK." - - if ether['IPv6'].src != proxy_ip: - raise RuntimeError("Source IP address error: {} != {}".format( - ether['IPv6'].src, proxy_ip)) - print "Source IP address: OK." - - if ether['IPv6'].dst != server_ip: - raise RuntimeError("Destination IP address error: {} != {}".format( - ether['IPv6'].dst, server_ip)) - print "Destination IP address: OK." - - msgtype = _get_dhcpv6_msgtype(ether['IPv6']['UDP'] - ['DHCPv6 Relay Forward Message (Relay Agent/Server Message)'].msgtype) - if msgtype != 'RELAY-FORW': - raise RuntimeError("Message type error: {} != RELAY-FORW".format( - msgtype)) - print "Message type: OK." - - linkaddr = ether['IPv6']['UDP']\ - ['DHCPv6 Relay Forward Message (Relay Agent/Server Message)'].linkaddr - if linkaddr != proxy_ip: - raise RuntimeError("Proxy IP address error: {} != {}".format( - linkaddr, proxy_ip)) - print "Proxy IP address: OK." - - try: - interface_id = ether['IPv6']['UDP']\ - ['DHCPv6 Relay Forward Message (Relay Agent/Server Message)']\ - ['Unknown DHCPv6 OPtion']['DHCP6 Interface-Id Option'].ifaceid - except Exception: - raise RuntimeError("DHCP6 Interface-Id error!") - - return interface_id - - -def dhcpv6_advertise(rx_if, tx_if, link_local_ip, proxy_ip, - server_ip, server_mac, proxy_to_server_mac, interface_id): - """Send and check DHCPv6 ADVERTISE proxy packet. - - :param rx_if: DHCPv6 server interface. - :param tx_if: Client interface. - :param link_local_ip: Client link-local address. - :param proxy_ip: IP address of DHCPv6 proxy server. - :param server_ip: IP address of DHCPv6 server. - :param server_mac: MAC address of DHCPv6 server. - :param proxy_to_server_mac: MAC address of DHCPv6 proxy interface. - :param interface_id: ID of proxy interface. - :type rx_if: str - :type tx_if: str - :type link_local_ip: str - :type proxy_ip: str - :type server_ip: str - :type server_mac: str - :type proxy_to_server_mac: str - :type interface_id: str - """ - - rxq = RxQueue(rx_if) - txq = TxQueue(tx_if) - - sent_packets = [] - - dhcp6_advertise_pkt = Ether(src=server_mac, dst=proxy_to_server_mac) / \ - IPv6(src=server_ip, dst=proxy_ip) / \ - UDP(sport=UDP_SERVICES.dhcpv6_server, - dport=UDP_SERVICES.dhcpv6_client) / \ - DHCP6_RelayReply(peeraddr=link_local_ip, - linkaddr=proxy_ip) / \ - DHCP6OptIfaceId(ifaceid=interface_id) / \ - DHCP6OptRelayMsg() / \ - DHCP6_Advertise() - - sent_packets.append(dhcp6_advertise_pkt) - txq.send(dhcp6_advertise_pkt) - - ether = rxq.recv(2) - - if ether is None: - raise RuntimeError('DHCPv6 ADVERTISE timeout') - - if ether['IPv6'].src != proxy_ip: - raise RuntimeError("Source IP address error: {} != {}".format( - ether['IPv6'].src, proxy_ip)) - print "Source IP address: OK." - - if not _check_udp_checksum(ether['IPv6']): - raise RuntimeError("Checksum error!") - print "Checksum: OK." - - msgtype = _get_dhcpv6_msgtype(ether['IPv6']['UDP'] - ['DHCPv6 Advertise Message'].msgtype) - if msgtype != 'ADVERTISE': - raise RuntimeError("Message type error: {} != ADVERTISE".format( - msgtype)) - print "Message type: OK." - - -def dhcpv6_request(tx_if, rx_if, dhcp_multicast_ip, link_local_ip, proxy_ip, - server_ip, client_duid, client_mac): - """Send and check DHCPv6 REQUEST proxy packet. - - :param tx_if: Client interface. - :param rx_if: DHCPv6 server interface. - :param dhcp_multicast_ip: Servers and relay agents multicast address. - :param link_local_ip: Client link-local address. - :param proxy_ip: IP address of DHCPv6 proxy server. - :param server_ip: IP address of DHCPv6 server. - :param client_duid: Client DHCP Unique Identifier. - :param client_mac: Client MAC address. - :type tx_if: str - :type rx_if: str - :type dhcp_multicast_ip: str - :type link_local_ip: str - :type proxy_ip: str - :type server_ip: str - :type client_duid: str - :type client_mac: str - """ - - rxq = RxQueue(rx_if) - txq = TxQueue(tx_if) - - sent_packets = [] - - dhcp6_request_pkt = Ether(src=client_mac, dst="33:33:00:01:00:02") / \ - IPv6(src=link_local_ip, dst=dhcp_multicast_ip) / \ - UDP(sport=UDP_SERVICES.dhcpv6_client, - dport=UDP_SERVICES.dhcpv6_server) / \ - DHCP6_Request() / \ - DHCP6OptClientId(duid=client_duid) - - sent_packets.append(dhcp6_request_pkt) - txq.send(dhcp6_request_pkt) - - ether = rxq.recv(2) - - if ether is None: - raise RuntimeError('DHCPv6 REQUEST timeout') - - if ether['IPv6'].src != proxy_ip: - raise RuntimeError("Source IP address error: {} != {}".format( - ether['IPv6'].src, proxy_ip)) - print "Source IP address: OK." - - if ether['IPv6'].dst != server_ip: - raise RuntimeError("Destination IP address error: {} != {}".format( - ether['IPv6'].dst, server_ip)) - print "Destination IP address: OK." - - msgtype = _get_dhcpv6_msgtype(ether['IPv6']['UDP'] - ['DHCPv6 Relay Forward Message (Relay Agent/Server Message)'].msgtype) - if msgtype != 'RELAY-FORW': - raise RuntimeError("Message type error: {} != RELAY-FORW".format( - msgtype)) - print "Message type: OK." - - linkaddr = ether['IPv6']['UDP']\ - ['DHCPv6 Relay Forward Message (Relay Agent/Server Message)'].linkaddr - if linkaddr != proxy_ip: - raise RuntimeError("Proxy IP address error: {} != {}".format( - linkaddr, proxy_ip)) - print "Proxy IP address: OK." - - -def dhcpv6_reply(rx_if, tx_if, link_local_ip, proxy_ip, server_ip, server_mac, - interface_id): - """Send and check DHCPv6 REPLY proxy packet. - - :param rx_if: DHCPv6 server interface. - :param tx_if: Client interface. - :param link_local_ip: Client link-local address. - :param proxy_ip: IP address of DHCPv6 proxy server. - :param server_ip: IP address of DHCPv6 server. - :param server_mac: MAC address of DHCPv6 server. - :param interface_id: ID of proxy interface. - :type rx_if: str - :type tx_if: str - :type link_local_ip: str - :type proxy_ip: str - :type server_ip: str - :type server_mac: str - :type interface_id: str - """ - - rxq = RxQueue(rx_if) - txq = TxQueue(tx_if) - - sent_packets = [] - - dhcp_reply_pkt = Ether(src=server_mac) / \ - IPv6(src=server_ip, dst=proxy_ip) / \ - UDP(sport=UDP_SERVICES.dhcpv6_server, - dport=UDP_SERVICES.dhcpv6_client) / \ - DHCP6_RelayReply(peeraddr=link_local_ip, - linkaddr=proxy_ip) / \ - DHCP6OptIfaceId(ifaceid=interface_id) / \ - DHCP6OptRelayMsg() / \ - DHCP6_Reply() - - sent_packets.append(dhcp_reply_pkt) - txq.send(dhcp_reply_pkt) - - ether = rxq.recv(2) - - if ether is None: - raise RuntimeError('DHCPv6 REPLY timeout') - - if ether['IPv6'].src != proxy_ip: - raise RuntimeError("Source IP address error: {} != {}".format( - ether['IPv6'].src, proxy_ip)) - print "Source IP address: OK." - - if not _check_udp_checksum(ether['IPv6']): - raise RuntimeError("Checksum error!") - print "Checksum: OK." - - msgtype = _get_dhcpv6_msgtype(ether['IPv6']['UDP'] - ['DHCPv6 Reply Message'].msgtype) - if msgtype != 'REPLY': - raise RuntimeError("Message type error: {} != REPLY".format(msgtype)) - print "Message type: OK." - - -def main(): - """Send DHCPv6 proxy messages.""" - - args = TrafficScriptArg(['tx_src_ip', 'tx_dst_ip', 'proxy_ip', 'proxy_mac', - 'server_ip', 'client_mac', 'server_mac', - 'proxy_to_server_mac']) - - client_if = args.get_arg('tx_if') - server_if = args.get_arg('rx_if') - proxy_ip = args.get_arg('proxy_ip') - proxy_mac = args.get_arg('proxy_mac') - proxy_to_server_mac = args.get_arg('proxy_to_server_mac') - server_ip = args.get_arg('server_ip') - client_mac = args.get_arg('client_mac') - server_mac = args.get_arg('server_mac') - - link_local_ip = "fe80::1" - dhcp_multicast_ip = "ff02::1:2" - client_duid = str(random.randint(0, 9999)) - - # SOLICIT - interface_id = dhcpv6_solicit(client_if, server_if, dhcp_multicast_ip, - link_local_ip, proxy_ip, server_ip, - server_mac, client_duid, client_mac) - - # ADVERTISE - dhcpv6_advertise(client_if, server_if, link_local_ip, proxy_ip, - server_ip, server_mac, proxy_to_server_mac, interface_id) - - # REQUEST - dhcpv6_request(client_if, server_if, dhcp_multicast_ip, link_local_ip, - proxy_ip, server_ip, client_duid, client_mac) - - # REPLY - dhcpv6_reply(client_if, server_if, link_local_ip, proxy_ip, server_ip, - server_mac, interface_id) - - sys.exit(0) - -if __name__ == "__main__": - main() |