diff options
Diffstat (limited to 'resources/traffic_scripts/dhcp')
6 files changed, 0 insertions, 1250 deletions
diff --git a/resources/traffic_scripts/dhcp/check_dhcp_discover.py b/resources/traffic_scripts/dhcp/check_dhcp_discover.py deleted file mode 100755 index 2fdc5b7fbf..0000000000 --- a/resources/traffic_scripts/dhcp/check_dhcp_discover.py +++ /dev/null @@ -1,150 +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 receives an DHCP packet on given interface and check if -is correct DHCP DISCOVER message. -""" - -import sys - -from scapy.layers.inet import UDP_SERVICES - -from resources.libraries.python.PacketVerifier import RxQueue -from resources.libraries.python.TrafficScriptArg import TrafficScriptArg - - -def main(): - args = TrafficScriptArg(['rx_src_mac'], ['hostname']) - - rx_if = args.get_arg('rx_if') - rx_src_mac = args.get_arg('rx_src_mac') - hostname = args.get_arg('hostname') - - rx_dst_mac = 'ff:ff:ff:ff:ff:ff' - rx_src_ip = '0.0.0.0' - rx_dst_ip = '255.255.255.255' - boot_request = 1 - dhcp_magic = 'c\x82Sc' - - rxq = RxQueue(rx_if) - - ether = rxq.recv(10) - - if ether is None: - raise RuntimeError("DHCP DISCOVER Rx timeout.") - - if ether.dst != rx_dst_mac: - raise RuntimeError("Destination MAC address error.") - print "Destination MAC address: OK." - - if ether.src != rx_src_mac: - raise RuntimeError("Source MAC address error.") - print "Source MAC address: OK." - - if ether['IP'].dst != rx_dst_ip: - raise RuntimeError("Destination IP address error.") - print "Destination IP address: OK." - - if ether['IP'].src != rx_src_ip: - raise RuntimeError("Source IP address error.") - print "Source IP address: OK." - - if ether['IP']['UDP'].dport != UDP_SERVICES.bootps: - raise RuntimeError("UDP destination port error.") - print "UDP destination port: OK." - - if ether['IP']['UDP'].sport != UDP_SERVICES.bootpc: - raise RuntimeError("UDP source port error.") - print "UDP source port: OK." - - bootp = ether['BOOTP'] - - if bootp.op != boot_request: - raise RuntimeError("BOOTP message type error.") - print "BOOTP message type: OK" - - if bootp.ciaddr != '0.0.0.0': - raise RuntimeError("BOOTP client IP address error.") - print "BOOTP client IP address: OK" - - if bootp.yiaddr != '0.0.0.0': - raise RuntimeError("BOOTP 'your' (client) IP address error.") - print "BOOTP 'your' (client) IP address: OK" - - if bootp.siaddr != '0.0.0.0': - raise RuntimeError("BOOTP next server IP address error.") - print "BOOTP next server IP address: OK" - - if bootp.giaddr != '0.0.0.0': - raise RuntimeError("BOOTP relay agent IP address error.") - print "BOOTP relay agent IP address: OK" - - chaddr = bootp.chaddr[:bootp.hlen].encode('hex') - if chaddr != rx_src_mac.replace(':', ''): - raise RuntimeError("BOOTP client hardware address error.") - print "BOOTP client hardware address: OK" - - # Check hostname - if bootp.sname != 64*'\x00': - raise RuntimeError("BOOTP server name error.") - print "BOOTP server name: OK" - - # Check boot file - if bootp.file != 128*'\x00': - raise RuntimeError("BOOTP boot file name error.") - print "BOOTP boot file name: OK" - - # Check bootp magic - if bootp.options != dhcp_magic: - raise RuntimeError("DHCP magic error.") - print "DHCP magic: OK" - - # Check options - dhcp_options = ether['DHCP options'].options - - # Option 12 - hn = filter(lambda x: x[0] == 'hostname', dhcp_options) - if hostname: - try: - if hn[0][1] != hostname: - raise RuntimeError("Client's hostname doesn't match.") - except IndexError: - raise RuntimeError("Option list doesn't contain hostname option.") - else: - if len(hn) != 0: - raise RuntimeError("Option list contains hostname option.") - print "Option 12 hostname: OK" - - # Option 53 - mt = filter(lambda x: x[0] == 'message-type', dhcp_options)[0][1] - if mt != 1: - raise RuntimeError("Option 53 message-type error.") - print "Option 53 message-type: OK" - - # Option 55 - prl = filter(lambda x: x[0] == 'param_req_list', dhcp_options)[0][1] - if prl != '\x01\x1c\x02\x03\x0f\x06w\x0c,/\x1ay*': - raise RuntimeError("Option 55 param_req_list error.") - print "Option 55 param_req_list: OK" - - # Option 255 - if 'end' not in dhcp_options: - raise RuntimeError("end option error.") - print "end option: OK" - - sys.exit(0) - - -if __name__ == "__main__": - main() diff --git a/resources/traffic_scripts/dhcp/check_dhcp_request.py b/resources/traffic_scripts/dhcp/check_dhcp_request.py deleted file mode 100755 index 522f2f507c..0000000000 --- a/resources/traffic_scripts/dhcp/check_dhcp_request.py +++ /dev/null @@ -1,218 +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 an DHCP OFFER message and checks if the DHCP -REQUEST contains all required fields.""" - -import sys - -from scapy.layers.l2 import Ether -from scapy.layers.inet import IP, UDP, UDP_SERVICES -from scapy.layers.dhcp import BOOTP, DHCP - -from resources.libraries.python.PacketVerifier import RxQueue, TxQueue -from resources.libraries.python.TrafficScriptArg import TrafficScriptArg - - -def is_discover(pkt): - """If DHCP message type option is set to dhcp discover return True, - else return False. False is returned also if exception occurs.""" - dhcp_discover = 1 - try: - dhcp_options = pkt['BOOTP']['DHCP options'].options - message_type = filter(lambda x: x[0] == 'message-type', - dhcp_options) - message_type = message_type[0][1] - return message_type == dhcp_discover - except: - return False - - -def is_request(pkt): - """If DHCP message type option is DHCP REQUEST return True, - else return False. False is returned also if exception occurs.""" - dhcp_request = 3 - try: - dhcp_options = pkt['BOOTP']['DHCP options'].options - message_type = filter(lambda x: x[0] == 'message-type', - dhcp_options) - message_type = message_type[0][1] - return message_type == dhcp_request - except: - return False - - -def main(): - """Main function of the script file.""" - args = TrafficScriptArg(['client_mac', 'server_mac', 'server_ip', - 'client_ip', 'client_mask'], - ['hostname', 'offer_xid']) - - server_if = args.get_arg('rx_if') - server_mac = args.get_arg('server_mac') - server_ip = args.get_arg('server_ip') - - client_mac = args.get_arg('client_mac') - client_ip = args.get_arg('client_ip') - client_mask = args.get_arg('client_mask') - - hostname = args.get_arg('hostname') - offer_xid = args.get_arg('offer_xid') - - rx_src_ip = '0.0.0.0' - rx_dst_ip = '255.255.255.255' - - rxq = RxQueue(server_if) - txq = TxQueue(server_if) - sent_packets = [] - - for _ in range(10): - dhcp_discover = rxq.recv(10) - if is_discover(dhcp_discover): - break - else: - raise RuntimeError("DHCP DISCOVER Rx error.") - - dhcp_offer = Ether(src=server_mac, dst=dhcp_discover.src) - dhcp_offer /= IP(src=server_ip, dst="255.255.255.255") - dhcp_offer /= UDP(sport=67, dport=68) - dhcp_offer /= BOOTP(op=2, - # if offer_xid differs from xid value in DHCP DISCOVER - # the DHCP OFFER has to be discarded - xid=int(offer_xid) if offer_xid - else dhcp_discover['BOOTP'].xid, - yiaddr=client_ip, - siaddr=server_ip, - chaddr=dhcp_discover['BOOTP'].chaddr) - dhcp_offer_options = [("message-type", "offer"), # Option 53 - ("subnet_mask", client_mask), # Option 1 - ("server_id", server_ip), # Option 54, dhcp server - ("lease_time", 43200), # Option 51 - "end"] - dhcp_offer /= DHCP(options=dhcp_offer_options) - - txq.send(dhcp_offer) - sent_packets.append(dhcp_offer) - - max_other_pkts = 10 - for _ in range(0, max_other_pkts): - dhcp_request = rxq.recv(5, sent_packets) - if not dhcp_request: - raise RuntimeError("DHCP REQUEST Rx timeout.") - if is_request(dhcp_request): - break - else: - raise RuntimeError("Max RX packet limit reached.") - - if offer_xid: - # if offer_xid differs from xid value in DHCP DISCOVER the DHCP OFFER - # has to be discarded - raise RuntimeError("DHCP REQUEST received. DHCP OFFER with wrong XID " - "has not been discarded.") - - # CHECK ETHER, IP, UDP - if dhcp_request.dst != dhcp_discover.dst: - raise RuntimeError("Destination MAC error.") - print "Destination MAC: OK." - - if dhcp_request.src != dhcp_discover.src: - raise RuntimeError("Source MAC error.") - print "Source MAC: OK." - - if dhcp_request['IP'].dst != rx_dst_ip: - raise RuntimeError("Destination IP error.") - print "Destination IP: OK." - - if dhcp_request['IP'].src != rx_src_ip: - raise RuntimeError("Source IP error.") - print "Source IP: OK." - - if dhcp_request['IP']['UDP'].dport != UDP_SERVICES.bootps: - raise RuntimeError("BOOTPs error.") - print "BOOTPs: OK." - - if dhcp_request['IP']['UDP'].sport != UDP_SERVICES.bootpc: - raise RuntimeError("BOOTPc error.") - print "BOOTPc: OK." - - # CHECK BOOTP - if dhcp_request['BOOTP'].op != dhcp_discover['BOOTP'].op: - raise RuntimeError("BOOTP operation error.") - print "BOOTP operation: OK" - - if dhcp_request['BOOTP'].xid != dhcp_discover['BOOTP'].xid: - raise RuntimeError("BOOTP XID error.") - print "BOOTP XID: OK" - - if dhcp_request['BOOTP'].ciaddr != '0.0.0.0': - raise RuntimeError("BOOTP ciaddr error.") - print "BOOTP ciaddr: OK" - - ca = dhcp_request['BOOTP'].chaddr[:dhcp_request['BOOTP'].hlen].encode('hex') - if ca != client_mac.replace(':', ''): - raise RuntimeError("BOOTP client hardware address error.") - print "BOOTP client hardware address: OK" - - if dhcp_request['BOOTP'].options != dhcp_discover['BOOTP'].options: - raise RuntimeError("DHCP options error.") - print "DHCP options: OK" - - # CHECK DHCP OPTIONS - dhcp_options = dhcp_request['DHCP options'].options - - hn = filter(lambda x: x[0] == 'hostname', dhcp_options) - if hostname: - try: - if hn[0][1] != hostname: - raise RuntimeError("Client's hostname doesn't match.") - except IndexError: - raise RuntimeError("Option list doesn't contain hostname option.") - else: - if len(hn) != 0: - raise RuntimeError("Option list contains hostname option.") - print "Option 12 hostname: OK" - - # Option 50 - ra = filter(lambda x: x[0] == 'requested_addr', dhcp_options)[0][1] - if ra != client_ip: - raise RuntimeError("Option 50 requested_addr error.") - print "Option 50 requested_addr: OK" - - # Option 53 - mt = filter(lambda x: x[0] == 'message-type', dhcp_options)[0][1] - if mt != 3: # request - raise RuntimeError("Option 53 message-type error.") - print "Option 53 message-type: OK" - - # Option 54 - sid = filter(lambda x: x[0] == 'server_id', dhcp_options)[0][1] - if sid != server_ip: - raise RuntimeError("Option 54 server_id error.") - print "Option 54 server_id: OK" - - # Option 55 - prl = filter(lambda x: x[0] == 'param_req_list', dhcp_options)[0][1] - if prl != '\x01\x1c\x02\x03\x0f\x06w\x0c,/\x1ay*': - raise RuntimeError("Option 55 param_req_list error.") - print "Option 55 param_req_list: OK" - - # Option 255 - if 'end' not in dhcp_options: - raise RuntimeError("end option error.") - print "end option: OK" - - sys.exit(0) - -if __name__ == "__main__": - main() diff --git a/resources/traffic_scripts/dhcp/check_dhcp_request_ack.py b/resources/traffic_scripts/dhcp/check_dhcp_request_ack.py deleted file mode 100755 index 8a3839cda7..0000000000 --- a/resources/traffic_scripts/dhcp/check_dhcp_request_ack.py +++ /dev/null @@ -1,132 +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 a DHCP ACK message when DHCP REQUEST message is -received.""" - -import sys - -from scapy.layers.l2 import Ether -from scapy.layers.inet import IP, UDP -from scapy.layers.dhcp import BOOTP, DHCP - -from resources.libraries.python.PacketVerifier import RxQueue, TxQueue -from resources.libraries.python.TrafficScriptArg import TrafficScriptArg - - -def is_discover(pkt): - """If DHCP message type option is set to dhcp discover return True, - else return False. False is returned also if exception occurs.""" - dhcp_discover = 1 - try: - dhcp_options = pkt['BOOTP']['DHCP options'].options - message_type = filter(lambda x: x[0] == 'message-type', - dhcp_options) - message_type = message_type[0][1] - return message_type == dhcp_discover - except: - return False - - -def is_request(pkt): - """If DHCP message type option is DHCP REQUEST return True, - else return False. False is returned also if exception occurs.""" - dhcp_request = 3 - try: - dhcp_options = pkt['BOOTP']['DHCP options'].options - message_type = filter(lambda x: x[0] == 'message-type', - dhcp_options) - message_type = message_type[0][1] - return message_type == dhcp_request - except: - return False - - -def main(): - """Main function of the script file.""" - args = TrafficScriptArg(['server_mac', 'server_ip', 'client_ip', - 'client_mask', 'lease_time']) - - server_if = args.get_arg('rx_if') - server_mac = args.get_arg('server_mac') - server_ip = args.get_arg('server_ip') - - client_ip = args.get_arg('client_ip') - client_mask = args.get_arg('client_mask') - - lease_time = int(args.get_arg('lease_time')) - - rxq = RxQueue(server_if) - txq = TxQueue(server_if) - sent_packets = [] - - for _ in range(10): - dhcp_discover = rxq.recv(10) - if is_discover(dhcp_discover): - break - else: - raise RuntimeError("DHCP DISCOVER Rx error.") - - dhcp_offer = Ether(src=server_mac, dst=dhcp_discover.src) - dhcp_offer /= IP(src=server_ip, dst="255.255.255.255") - dhcp_offer /= UDP(sport=67, dport=68) - dhcp_offer /= BOOTP(op=2, - xid=dhcp_discover['BOOTP'].xid, - yiaddr=client_ip, - siaddr=server_ip, - chaddr=dhcp_discover['BOOTP'].chaddr) - dhcp_offer_options = [("message-type", "offer"), # Option 53 - ("subnet_mask", client_mask), # Option 1 - ("server_id", server_ip), # Option 54, dhcp server - ("lease_time", lease_time), # Option 51 - "end"] - dhcp_offer /= DHCP(options=dhcp_offer_options) - - txq.send(dhcp_offer) - sent_packets.append(dhcp_offer) - - max_other_pkts = 10 - for _ in range(0, max_other_pkts): - dhcp_request = rxq.recv(5, sent_packets) - if not dhcp_request: - raise RuntimeError("DHCP REQUEST Rx timeout.") - if is_request(dhcp_request): - break - else: - raise RuntimeError("Max RX packet limit reached.") - - # Send dhcp ack - dhcp_ack = Ether(src=server_mac, dst=dhcp_request.src) - dhcp_ack /= IP(src=server_ip, dst="255.255.255.255") - dhcp_ack /= UDP(sport=67, dport=68) - dhcp_ack /= BOOTP(op=2, - xid=dhcp_request['BOOTP'].xid, - yiaddr=client_ip, - siaddr=server_ip, - flags=dhcp_request['BOOTP'].flags, - chaddr=dhcp_request['BOOTP'].chaddr) - dhcp_ack_options = [("message-type", "ack"), # Option 53. 5: ACK, 6: NAK - ("subnet_mask", client_mask), # Option 1 - ("server_id", server_ip), # Option 54, dhcp server - ("lease_time", lease_time), # Option 51, - "end"] - dhcp_ack /= DHCP(options=dhcp_ack_options) - - txq.send(dhcp_ack) - sent_packets.append(dhcp_ack) - - sys.exit(0) - -if __name__ == "__main__": - main() diff --git a/resources/traffic_scripts/dhcp/send_and_check_proxy_discover.py b/resources/traffic_scripts/dhcp/send_and_check_proxy_discover.py deleted file mode 100755 index d8089713fd..0000000000 --- a/resources/traffic_scripts/dhcp/send_and_check_proxy_discover.py +++ /dev/null @@ -1,79 +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 DHCP DISCOVER packet - and check if is received on interface.""" - -import sys - -from scapy.all import Ether -from scapy.layers.inet import IP, UDP -from scapy.layers.inet import UDP_SERVICES -from scapy.layers.dhcp import DHCP, BOOTP - -from resources.libraries.python.PacketVerifier import RxQueue, TxQueue -from resources.libraries.python.TrafficScriptArg import TrafficScriptArg - - -def is_discover(pkt): - """If DHCP message type option is set to dhcp discover return True, - else return False. False is returned also if exception occurs.""" - dhcp_discover = 1 - try: - dhcp_options = pkt['BOOTP']['DHCP options'].options - message_type = filter(lambda x: x[0] == 'message-type', - dhcp_options) - message_type = message_type[0][1] - return message_type == dhcp_discover - except: - return False - - -def main(): - """Send DHCP DISCOVER packet.""" - - args = TrafficScriptArg(['tx_src_ip', 'tx_dst_ip']) - - tx_if = args.get_arg('tx_if') - rx_if = args.get_arg('rx_if') - - rxq = RxQueue(rx_if) - txq = TxQueue(tx_if) - - tx_src_ip = args.get_arg('tx_src_ip') - tx_dst_ip = args.get_arg('tx_dst_ip') - - sent_packets = [] - - dhcp_discover = Ether(dst="ff:ff:ff:ff:ff:ff") / \ - IP(src=tx_src_ip, dst=tx_dst_ip) / \ - UDP(sport=UDP_SERVICES.bootpc, dport=UDP_SERVICES.bootps) / \ - BOOTP(op=1,) / \ - DHCP(options=[("message-type", "discover"), - "end"]) - - sent_packets.append(dhcp_discover) - txq.send(dhcp_discover) - - for _ in range(10): - dhcp_discover = rxq.recv(2) - if is_discover(dhcp_discover): - break - else: - raise RuntimeError("DHCP DISCOVER Rx timeout") - - sys.exit(0) - -if __name__ == "__main__": - main() diff --git a/resources/traffic_scripts/dhcp/send_and_check_proxy_messages.py b/resources/traffic_scripts/dhcp/send_and_check_proxy_messages.py deleted file mode 100755 index 27f148c900..0000000000 --- a/resources/traffic_scripts/dhcp/send_and_check_proxy_messages.py +++ /dev/null @@ -1,299 +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 DHCP packets.""" - -import sys - -from scapy.all import Ether -from scapy.layers.inet import IP, UDP -from scapy.layers.inet import UDP_SERVICES -from scapy.layers.dhcp import DHCP, BOOTP - -from resources.libraries.python.PacketVerifier import RxQueue, TxQueue -from resources.libraries.python.TrafficScriptArg import TrafficScriptArg - - -def dhcp_discover(tx_if, rx_if, tx_src_ip, tx_dst_ip, server_ip, proxy_ip, - client_mac): - """Send and check DHCP DISCOVER proxy packet.""" - - rxq = RxQueue(rx_if) - txq = TxQueue(tx_if) - - sent_packets = [] - - dhcp_discover = Ether(src=client_mac, dst="ff:ff:ff:ff:ff:ff") / \ - IP(src=tx_src_ip, dst=tx_dst_ip) / \ - UDP(sport=UDP_SERVICES.bootpc, dport=UDP_SERVICES.bootps) / \ - BOOTP(op=1,) / \ - DHCP(options=[("message-type", "discover"), - "end"]) - - sent_packets.append(dhcp_discover) - txq.send(dhcp_discover) - - ether = rxq.recv(2) - - if ether is None: - raise RuntimeError('DHCP DISCOVER timeout') - - if ether[IP].src != proxy_ip: - raise RuntimeError("Source IP address error.") - print "Source IP address: OK." - - if ether[IP].dst != server_ip: - raise RuntimeError("Destination IP address error.") - print "Destination IP address: OK." - - if ether[UDP].dport != UDP_SERVICES.bootps: - raise RuntimeError("UDP destination port error.") - print "UDP destination port: OK." - - if ether[UDP].sport != UDP_SERVICES.bootpc: - raise RuntimeError("UDP source port error.") - print "UDP source port: OK." - - if ether[DHCP].options[1][0] != 'relay_agent_Information': # option 82 - raise RuntimeError("Relay agent information error.") - option_82 = ether[DHCP].options[1][1] - - if ether[DHCP].options[0][1] != 1: # 1 - DISCOVER message - raise RuntimeError("DHCP DISCOVER message error.") - print "DHCP DISCOVER message OK." - - return option_82 - - -def dhcp_offer(rx_if, tx_if, tx_dst_ip, server_ip, proxy_ip, client_ip, - server_mac, option_82): - """Send and check DHCP OFFER proxy packet.""" - - rxq = RxQueue(rx_if) - txq = TxQueue(tx_if) - - sent_packets = [] - - dhcp_offer = Ether(src=server_mac, dst="ff:ff:ff:ff:ff:ff") / \ - IP(src=server_ip, dst=tx_dst_ip) / \ - UDP(sport=UDP_SERVICES.bootps, dport=UDP_SERVICES.bootpc) / \ - BOOTP(op=2, - yiaddr=client_ip, - siaddr=server_ip) / \ - DHCP(options= - [("message-type", "offer"), - ("server_id", server_ip), - ("relay_agent_Information", option_82), - "end"]) - - txq.send(dhcp_offer) - sent_packets.append(dhcp_offer) - - ether = rxq.recv(2) - - if ether is None: - raise RuntimeError('DHCP OFFER timeout') - - if ether[IP].dst != tx_dst_ip: - raise RuntimeError("Destination IP address error.") - print "Destination IP address: OK." - - if ether[IP].src != proxy_ip: - raise RuntimeError("Source IP address error.") - print "Source IP address: OK." - - if ether[UDP].dport != UDP_SERVICES.bootpc: - raise RuntimeError("UDP destination port error.") - print "UDP destination port: OK." - - if ether[UDP].sport != UDP_SERVICES.bootps: - raise RuntimeError("UDP source port error.") - print "UDP source port: OK." - - if ether[BOOTP].yiaddr != client_ip: - raise RuntimeError("Client IP address error.") - print "Client IP address: OK." - - if ether[BOOTP].siaddr != server_ip: - raise RuntimeError("DHCP server IP address error.") - print "DHCP server IP address: OK." - - if ether[DHCP].options[0][1] != 2: # 2 - OFFER message - raise RuntimeError("DHCP OFFER message error.") - print "DHCP OFFER message OK." - - -def dhcp_request(tx_if, rx_if, tx_src_ip, tx_dst_ip, server_ip, proxy_ip, - client_ip, client_mac): - """Send and check DHCP REQUEST proxy packet.""" - - rxq = RxQueue(rx_if) - txq = TxQueue(tx_if) - - sent_packets = [] - - dhcp_request = Ether(src=client_mac, dst="ff:ff:ff:ff:ff:ff") / \ - IP(src=tx_src_ip, dst=tx_dst_ip) / \ - UDP(sport=UDP_SERVICES.bootpc, dport=UDP_SERVICES.bootps) / \ - BOOTP(op=1, - giaddr=proxy_ip, - siaddr=server_ip) / \ - DHCP(options=[("message-type", "request"), - ("server_id", server_ip), - ("requested_addr", client_ip), - "end"]) - - sent_packets.append(dhcp_request) - txq.send(dhcp_request) - - ether = rxq.recv(2) - - if ether is None: - raise RuntimeError('DHCP REQUEST timeout') - - if ether[IP].dst != server_ip: - raise RuntimeError("Destination IP address error.") - print "Destination IP address: OK." - - if ether[IP].src != proxy_ip: - raise RuntimeError("Source IP address error.") - print "Source IP address: OK." - - if ether[UDP].dport != UDP_SERVICES.bootps: - raise RuntimeError("UDP destination port error.") - print "UDP destination port: OK." - - if ether[UDP].sport != UDP_SERVICES.bootpc: - raise RuntimeError("UDP source port error.") - print "UDP source port: OK." - - if ether[BOOTP].siaddr != server_ip: - raise RuntimeError("DHCP server IP address error.") - print "DHCP server IP address: OK." - - if ether[DHCP].options[2][1] != client_ip: - raise RuntimeError("Requested IP address error.") - print "Requested IP address: OK." - - if ether[DHCP].options[3][0] != 'relay_agent_Information': # option 82 - raise RuntimeError("Relay agent information error.") - - if ether[DHCP].options[0][1] != 3: # 2 - REQUEST message - raise RuntimeError("DHCP REQUEST message error.") - print "DHCP REQUEST message: OK." - - -def dhcp_ack(rx_if, tx_if, tx_dst_ip, server_ip, proxy_ip, client_ip, - server_mac, option_82): - """Send and check DHCP ACK proxy packet.""" - - rxq = RxQueue(rx_if) - txq = TxQueue(tx_if) - - lease_time = 43200 # 12 hours - - sent_packets = [] - - dhcp_ack = Ether(src=server_mac, dst="ff:ff:ff:ff:ff:ff") / \ - IP(src=server_ip, dst=tx_dst_ip) / \ - UDP(sport=UDP_SERVICES.bootps, dport=UDP_SERVICES.bootpc) / \ - BOOTP(op=2, - yiaddr=client_ip, - siaddr=server_ip) / \ - DHCP(options= - [("message-type", "ack"), - ("server_id", server_ip), - ("lease_time", lease_time), - ("relay_agent_Information", option_82), - "end"]) - - txq.send(dhcp_ack) - sent_packets.append(dhcp_ack) - - ether = rxq.recv(2) - - if ether is None: - raise RuntimeError('DHCP ACK timeout') - - if ether[IP].dst != tx_dst_ip: - raise RuntimeError("Destination IP address error.") - print "Destination IP address: OK." - - if ether[IP].src != proxy_ip: - raise RuntimeError("Source IP address error.") - print "Source IP address: OK." - - if ether[UDP].dport != UDP_SERVICES.bootpc: - raise RuntimeError("UDP destination port error.") - print "UDP destination port: OK." - - if ether[UDP].sport != UDP_SERVICES.bootps: - raise RuntimeError("UDP source port error.") - print "UDP source port: OK." - - if ether[BOOTP].yiaddr != client_ip: - raise RuntimeError("Client IP address error.") - print "Client IP address: OK." - - if ether[BOOTP].siaddr != server_ip: - raise RuntimeError("DHCP server IP address error.") - print "DHCP server IP address: OK." - - if ether[DHCP].options[2][1] != lease_time: - raise RuntimeError("DHCP lease time error.") - print "DHCP lease time OK." - - if ether[DHCP].options[0][1] != 5: # 5 - ACK message - raise RuntimeError("DHCP ACK message error.") - print "DHCP ACK message OK." - - -def main(): - """Send DHCP proxy messages.""" - - args = TrafficScriptArg(['server_ip', 'server_mac', 'client_ip', - 'client_mac', 'proxy_ip']) - - tx_if = args.get_arg('tx_if') - rx_if = args.get_arg('rx_if') - - tx_src_ip = "0.0.0.0" - tx_dst_ip = "255.255.255.255" - - server_ip = args.get_arg('server_ip') - client_ip = args.get_arg('client_ip') - proxy_ip = args.get_arg('proxy_ip') - client_mac = args.get_arg('client_mac') - server_mac = args.get_arg('server_mac') - - # DHCP DISCOVER - option_82 = dhcp_discover(tx_if, rx_if, tx_src_ip, tx_dst_ip, server_ip, - proxy_ip, client_mac) - - # DHCP OFFER - dhcp_offer(tx_if, rx_if, tx_dst_ip, server_ip, proxy_ip, client_ip, - server_mac, option_82) - - # DHCP REQUEST - dhcp_request(tx_if, rx_if, tx_src_ip, tx_dst_ip, server_ip, proxy_ip, - client_ip, client_mac) - - # DHCP ACK - dhcp_ack(tx_if, rx_if, tx_dst_ip, server_ip, proxy_ip, client_ip, - server_mac, option_82) - - sys.exit(0) - -if __name__ == "__main__": - main() 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() |