diff options
Diffstat (limited to 'resources/traffic_scripts')
| -rwxr-xr-x | resources/traffic_scripts/icmpv6_echo.py | 99 | ||||
| -rwxr-xr-x | resources/traffic_scripts/icmpv6_echo_req_resp.py | 185 | ||||
| -rwxr-xr-x | resources/traffic_scripts/ipv4_ping_ttl_check.py | 124 | ||||
| -rwxr-xr-x | resources/traffic_scripts/ipv6_ns.py | 104 | ||||
| -rwxr-xr-x | resources/traffic_scripts/ipv6_sweep_ping.py | 110 | ||||
| -rwxr-xr-x | resources/traffic_scripts/send_ip_icmp.py | 71 |
6 files changed, 693 insertions, 0 deletions
diff --git a/resources/traffic_scripts/icmpv6_echo.py b/resources/traffic_scripts/icmpv6_echo.py new file mode 100755 index 0000000000..c3c8d5a381 --- /dev/null +++ b/resources/traffic_scripts/icmpv6_echo.py @@ -0,0 +1,99 @@ +#!/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 for ICMPv6 echo test.""" + +import sys +import logging +logging.getLogger("scapy.runtime").setLevel(logging.ERROR) +from resources.libraries.python.PacketVerifier import RxQueue, TxQueue +from resources.libraries.python.TrafficScriptArg import TrafficScriptArg +from scapy.layers.inet6 import IPv6, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr +from scapy.layers.inet6 import ICMPv6EchoRequest, ICMPv6EchoReply +from scapy.all import Ether + + +def main(): + args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip']) + + rxq = RxQueue(args.get_arg('rx_if')) + txq = TxQueue(args.get_arg('tx_if')) + + src_mac = args.get_arg('src_mac') + dst_mac = args.get_arg('dst_mac') + src_ip = args.get_arg('src_ip') + dst_ip = args.get_arg('dst_ip') + echo_id = 0xa + echo_seq = 0x1 + + sent_packets = [] + + # send ICMPv6 neighbor advertisement message + pkt_send = (Ether(src=src_mac, dst='ff:ff:ff:ff:ff:ff') / + IPv6(src=src_ip, dst='ff02::1:ff00:2') / + ICMPv6ND_NA(tgt=src_ip, R=0) / + ICMPv6NDOptDstLLAddr(lladdr=src_mac)) + sent_packets.append(pkt_send) + txq.send(pkt_send) + + # send ICMPv6 echo request + pkt_send = (Ether(src=src_mac, dst=dst_mac) / + IPv6(src=src_ip, dst=dst_ip) / + ICMPv6EchoRequest(id=echo_id, seq=echo_seq)) + sent_packets.append(pkt_send) + txq.send(pkt_send) + + # receive ICMPv6 echo reply + ether = rxq.recv(2, sent_packets) + if ether is None: + rxq._proc.terminate() + raise RuntimeError('ICMPv6 echo reply Rx timeout') + + if not ether.haslayer(IPv6): + rxq._proc.terminate() + raise RuntimeError('Unexpected packet with no IPv6 received {0}'.format( + ether.__repr__())) + + ipv6 = ether['IPv6'] + + if not ipv6.haslayer(ICMPv6EchoReply): + rxq._proc.terminate() + raise RuntimeError( + 'Unexpected packet with no IPv6 ICMP received {0}'.format( + ipv6.__repr__())) + + icmpv6 = ipv6['ICMPv6 Echo Reply'] + + # check identifier and sequence number + if icmpv6.id != echo_id or icmpv6.seq != echo_seq: + rxq._proc.terminate() + raise RuntimeError( + 'Invalid ICMPv6 echo reply received ID {0} seq {1} should be ' + + 'ID {2} seq {3}'.format(icmpv6.id, icmpv6.seq, echo_id, echo_seq)) + + # verify checksum + cksum = icmpv6.cksum + del icmpv6.cksum + tmp = ICMPv6EchoReply(str(icmpv6)) + if tmp.cksum != cksum: + rxq._proc.terminate() + raise RuntimeError( + 'Invalid checksum {0} should be {1}'.format(cksum, tmp.cksum)) + + rxq._proc.terminate() + sys.exit(0) + +if __name__ == "__main__": + main() diff --git a/resources/traffic_scripts/icmpv6_echo_req_resp.py b/resources/traffic_scripts/icmpv6_echo_req_resp.py new file mode 100755 index 0000000000..24f4faa3f4 --- /dev/null +++ b/resources/traffic_scripts/icmpv6_echo_req_resp.py @@ -0,0 +1,185 @@ +#!/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. + +"""Send ICMPv6 echo request from one TG port to another through DUT nodes and + send reply back. Also verify hop limit processing.""" + +import sys +import logging +logging.getLogger("scapy.runtime").setLevel(logging.ERROR) +from resources.libraries.python.PacketVerifier import RxQueue, TxQueue +from resources.libraries.python.TrafficScriptArg import TrafficScriptArg +from scapy.layers.inet6 import IPv6, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr +from scapy.layers.inet6 import ICMPv6EchoRequest, ICMPv6EchoReply +from scapy.all import Ether + + +def main(): + args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_nh_mac', 'dst_nh_mac', + 'src_ip', 'dst_ip', 'h_num']) + + src_rxq = RxQueue(args.get_arg('rx_if')) + src_txq = TxQueue(args.get_arg('rx_if')) + dst_rxq = RxQueue(args.get_arg('tx_if')) + dst_txq = TxQueue(args.get_arg('tx_if')) + + src_mac = args.get_arg('src_mac') + dst_mac = args.get_arg('dst_mac') + src_nh_mac = args.get_arg('src_nh_mac') + dst_nh_mac = args.get_arg('dst_nh_mac') + src_ip = args.get_arg('src_ip') + dst_ip = args.get_arg('dst_ip') + hop_num = int(args.get_arg('h_num')) + hop_limit = 64 + echo_id = 0xa + echo_seq = 0x1 + + src_sent_packets = [] + dst_sent_packets = [] + + # send ICMPv6 neighbor advertisement message + pkt_send = (Ether(src=src_mac, dst='ff:ff:ff:ff:ff:ff') / + IPv6(src=src_ip, dst='ff02::1:ff00:2') / + ICMPv6ND_NA(tgt=src_ip, R=0) / + ICMPv6NDOptDstLLAddr(lladdr=src_mac)) + src_sent_packets.append(pkt_send) + src_txq.send(pkt_send) + pkt_send = (Ether(src=dst_mac, dst='ff:ff:ff:ff:ff:ff') / + IPv6(src=dst_ip, dst='ff02::1:ff00:2') / + ICMPv6ND_NA(tgt=dst_ip, R=0) / + ICMPv6NDOptDstLLAddr(lladdr=dst_mac)) + dst_sent_packets.append(pkt_send) + dst_txq.send(pkt_send) + + # send ICMPv6 echo request from first TG interface + pkt_send = (Ether(src=src_mac, dst=src_nh_mac) / + IPv6(src=src_ip, dst=dst_ip, hlim=hop_limit) / + ICMPv6EchoRequest(id=echo_id, seq=echo_seq)) + src_sent_packets.append(pkt_send) + src_txq.send(pkt_send) + + # receive ICMPv6 echo request on second TG interface + ether = dst_rxq.recv(2, dst_sent_packets) + if ether is None: + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError('ICMPv6 echo reply Rx timeout') + + if not ether.haslayer(IPv6): + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError('Unexpected packet with no IPv6 received {0}'.format( + ether.__repr__())) + + ipv6 = ether['IPv6'] + + # verify hop limit processing + if ipv6.hlim != (hop_limit - hop_num): + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError( + 'Invalid hop limit {0} should be {1}'.format(ipv6.hlim, + hop_limit - hop_num)) + + if not ipv6.haslayer(ICMPv6EchoRequest): + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError( + 'Unexpected packet with no IPv6 ICMP received {0}'.format( + ipv6.__repr__())) + + icmpv6 = ipv6['ICMPv6 Echo Request'] + + # check identifier and sequence number + if icmpv6.id != echo_id or icmpv6.seq != echo_seq: + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError( + 'Invalid ICMPv6 echo reply received ID {0} seq {1} should be ' + + 'ID {2} seq {3}'.format(icmpv6.id, icmpv6.seq, echo_id, echo_seq)) + + # verify checksum + cksum = icmpv6.cksum + del icmpv6.cksum + tmp = ICMPv6EchoRequest(str(icmpv6)) + if tmp.cksum != cksum: + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError( + 'Invalid checksum {0} should be {1}'.format(cksum, tmp.cksum)) + + # send ICMPv6 echo reply from second TG interface + pkt_send = (Ether(src=dst_mac, dst=dst_nh_mac) / + IPv6(src=dst_ip, dst=src_ip) / + ICMPv6EchoReply(id=echo_id, seq=echo_seq)) + dst_sent_packets.append(pkt_send) + dst_txq.send(pkt_send) + + # receive ICMPv6 echo reply on first TG interface + ether = src_rxq.recv(2, src_sent_packets) + if ether is None: + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError('ICMPv6 echo reply Rx timeout') + + if not ether.haslayer(IPv6): + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError('Unexpected packet with no IPv6 received {0}'.format( + ether.__repr__())) + + ipv6 = ether['IPv6'] + + # verify hop limit processing + if ipv6.hlim != (hop_limit - hop_num): + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError( + 'Invalid hop limit {0} should be {1}'.format(ipv6.hlim, + hop_limit - hop_num)) + + if not ipv6.haslayer(ICMPv6EchoReply): + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError( + 'Unexpected packet with no IPv6 ICMP received {0}'.format( + ipv6.__repr__())) + + icmpv6 = ipv6['ICMPv6 Echo Reply'] + + # check identifier and sequence number + if icmpv6.id != echo_id or icmpv6.seq != echo_seq: + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError( + 'Invalid ICMPv6 echo reply received ID {0} seq {1} should be ' + + 'ID {2} seq {3}'.format(icmpv6.id, icmpv6.seq, echo_id, echo_seq)) + + # verify checksum + cksum = icmpv6.cksum + del icmpv6.cksum + tmp = ICMPv6EchoReply(str(icmpv6)) + if tmp.cksum != cksum: + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + raise RuntimeError( + 'Invalid checksum {0} should be {1}'.format(cksum, tmp.cksum)) + + src_rxq._proc.terminate() + dst_rxq._proc.terminate() + sys.exit(0) + +if __name__ == "__main__": + main() diff --git a/resources/traffic_scripts/ipv4_ping_ttl_check.py b/resources/traffic_scripts/ipv4_ping_ttl_check.py new file mode 100755 index 0000000000..050a1d7b29 --- /dev/null +++ b/resources/traffic_scripts/ipv4_ping_ttl_check.py @@ -0,0 +1,124 @@ +#!/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. + +from scapy.all import * +from resources.libraries.python.PacketVerifier \ + import Interface, create_gratuitous_arp_request, auto_pad +from optparse import OptionParser + + +def check_ttl(ttl_begin, ttl_end, ttl_diff): + if ttl_begin != ttl_end + ttl_diff: + src_if.close() + if dst_if_defined: + dst_if.close() + raise Exception( + "TTL changed from {} to {} but decrease by {} expected")\ + .format(ttl_begin, ttl_end, hops) + + +def ckeck_packets_equal(pkt_send, pkt_recv): + pkt_send_raw = str(pkt_send) + pkt_recv_raw = str(pkt_recv) + if pkt_send_raw != pkt_recv_raw: + print "Sent: {}".format(pkt_send_raw.encode('hex')) + print "Received: {}".format(pkt_recv_raw.encode('hex')) + print "Sent:" + Ether(pkt_send_raw).show2() + print "Received:" + Ether(pkt_recv_raw).show2() + src_if.close() + if dst_if_defined: + dst_if.close() + raise Exception("Sent packet doesn't match received packet") + + +parser = OptionParser() +parser.add_option("--src_if", dest="src_if") +parser.add_option("--dst_if", dest="dst_if") # optional +parser.add_option("--src_mac", dest="src_mac") +parser.add_option("--first_hop_mac", dest="first_hop_mac") +parser.add_option("--dst_mac", dest="dst_mac") # optional +parser.add_option("--src_ip", dest="src_ip") +parser.add_option("--dst_ip", dest="dst_ip") +parser.add_option("--hops", dest="hops") # optional +# If one of 'dst_if', 'dst_mac' and 'hops' is specified all must be specified. +(opts, args) = parser.parse_args() +src_if_name = opts.src_if +dst_if_name = opts.dst_if +dst_if_defined = True +if dst_if_name is None: + dst_if_defined = False +src_mac = opts.src_mac +first_hop_mac = opts.first_hop_mac +dst_mac = opts.dst_mac +src_ip = opts.src_ip +dst_ip = opts.dst_ip +hops = int(opts.hops) + +if dst_if_defined and (src_if_name == dst_if_name): + raise Exception("Source interface name equals destination interface name") + +src_if = Interface(src_if_name) +src_if.send_pkt(create_gratuitous_arp_request(src_mac, src_ip)) +if dst_if_defined: + dst_if = Interface(dst_if_name) + dst_if.send_pkt(create_gratuitous_arp_request(dst_mac, dst_ip)) + +pkt_req_send = auto_pad(Ether(src=src_mac, dst=first_hop_mac) / + IP(src=src_ip, dst=dst_ip) / + ICMP()) +pkt_req_send = Ether(pkt_req_send) +src_if.send_pkt(pkt_req_send) + +if dst_if_defined: + try: + pkt_req_recv = dst_if.recv_pkt() + except: + src_if.close() + if dst_if_defined: + dst_if.close() + raise + + check_ttl(pkt_req_send[IP].ttl, pkt_req_recv[IP].ttl, hops) + pkt_req_send_mod = pkt_req_send.copy() + pkt_req_send_mod[IP].ttl = pkt_req_recv[IP].ttl + del pkt_req_send_mod[IP].chksum # update checksum + ckeck_packets_equal(pkt_req_send_mod[IP], pkt_req_recv[IP]) + + pkt_resp_send = auto_pad(Ether(src=dst_mac, dst=pkt_req_recv.src) / + IP(src=dst_ip, dst=src_ip) / + ICMP(type=0)) # echo-reply + pkt_resp_send = Ether(pkt_resp_send) + dst_if.send_pkt(pkt_resp_send) + +try: + pkt_resp_recv = src_if.recv_pkt() +except: + src_if.close() + if dst_if_defined: + dst_if.close() + raise + +if dst_if_defined: + check_ttl(pkt_resp_send[IP].ttl, pkt_resp_recv[IP].ttl, hops) + pkt_resp_send_mod = pkt_resp_send.copy() + pkt_resp_send_mod[IP].ttl = pkt_resp_recv[IP].ttl + del pkt_resp_send_mod[IP].chksum # update checksum + ckeck_packets_equal(pkt_resp_send_mod[IP], pkt_resp_recv[IP]) + +src_if.close() +if dst_if_defined: + dst_if.close() diff --git a/resources/traffic_scripts/ipv6_ns.py b/resources/traffic_scripts/ipv6_ns.py new file mode 100755 index 0000000000..dd1adad39e --- /dev/null +++ b/resources/traffic_scripts/ipv6_ns.py @@ -0,0 +1,104 @@ +#!/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 for IPv6 Neighbor Solicitation test.""" + +import sys +import logging +logging.getLogger("scapy.runtime").setLevel(logging.ERROR) +from resources.libraries.python.PacketVerifier import RxQueue, TxQueue +from resources.libraries.python.TrafficScriptArg import TrafficScriptArg +from scapy.layers.inet6 import IPv6, ICMPv6ND_NA, ICMPv6ND_NS +from scapy.layers.inet6 import ICMPv6NDOptDstLLAddr, ICMPv6NDOptSrcLLAddr +from scapy.all import Ether + + +def main(): + args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip']) + + rxq = RxQueue(args.get_arg('rx_if')) + txq = TxQueue(args.get_arg('tx_if')) + + src_mac = args.get_arg('src_mac') + dst_mac = args.get_arg('dst_mac') + src_ip = args.get_arg('src_ip') + dst_ip = args.get_arg('dst_ip') + + sent_packets = [] + + # send ICMPv6 neighbor solicitation message + pkt_send = (Ether(src=src_mac, dst='ff:ff:ff:ff:ff:ff') / + IPv6(src=src_ip, dst='ff02::1:ff00:2') / + ICMPv6ND_NS(tgt=dst_ip) / + ICMPv6NDOptSrcLLAddr(lladdr=src_mac)) + sent_packets.append(pkt_send) + txq.send(pkt_send) + + # receive ICMPv6 neighbor advertisement message + ether = rxq.recv(2, sent_packets) + if ether is None: + rxq._proc.terminate() + raise RuntimeError('ICMPv6 echo reply Rx timeout') + + if not ether.haslayer(IPv6): + rxq._proc.terminate() + raise RuntimeError('Unexpected packet with no IPv6 received {0}'.format( + ether.__repr__())) + + ipv6 = ether['IPv6'] + + if not ipv6.haslayer(ICMPv6ND_NA): + rxq._proc.terminate() + raise RuntimeError( + 'Unexpected packet with no ICMPv6 ND-NA received {0}'.format( + ipv6.__repr__())) + + icmpv6_na = ipv6['ICMPv6 Neighbor Discovery - Neighbor Advertisement'] + + # verify target address + if icmpv6_na.tgt != dst_ip: + rxq._proc.terminate() + raise RuntimeError('Invalid target address {0} should be {1}'.format( + icmpv6_na.tgt, dst_ip)) + + if not icmpv6_na.haslayer(ICMPv6NDOptDstLLAddr): + rxq._proc.terminate() + raise RuntimeError( + 'Missing Destination Link-Layer Address option in ICMPv6 ' + + 'Neighbor Advertisement {0}'.format(icmpv6_na.__repr__())) + + option = 'ICMPv6 Neighbor Discovery Option - Destination Link-Layer Address' + dst_ll_addr = icmpv6_na[option] + + # verify destination link-layer address field + if dst_ll_addr.lladdr != dst_mac: + rxq._proc.terminate() + raise RuntimeError('Invalid lladdr {0} should be {1}'.format( + dst_ll_addr.lladdr, dst_mac)) + + # verify checksum + cksum = icmpv6_na.cksum + del icmpv6_na.cksum + tmp = ICMPv6ND_NA(str(icmpv6_na)) + if tmp.cksum != cksum: + rxq._proc.terminate() + raise RuntimeError( + 'Invalid checksum {0} should be {1}'.format(cksum, tmp.cksum)) + + rxq._proc.terminate() + sys.exit(0) + +if __name__ == "__main__": + main() diff --git a/resources/traffic_scripts/ipv6_sweep_ping.py b/resources/traffic_scripts/ipv6_sweep_ping.py new file mode 100755 index 0000000000..2282f40f78 --- /dev/null +++ b/resources/traffic_scripts/ipv6_sweep_ping.py @@ -0,0 +1,110 @@ +#!/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 for IPv6 sweep ping.""" + +import sys +import logging +import os +logging.getLogger("scapy.runtime").setLevel(logging.ERROR) +from resources.libraries.python.PacketVerifier import RxQueue, TxQueue +from resources.libraries.python.TrafficScriptArg import TrafficScriptArg +from scapy.layers.inet6 import IPv6, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr +from scapy.layers.inet6 import ICMPv6EchoRequest, ICMPv6EchoReply +from scapy.all import Ether + + +def main(): + # start_size - start size of the ICMPv6 echo data + # end_size - end size of the ICMPv6 echo data + # step - increment step + args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip', + 'start_size', 'end_size', 'step']) + + rxq = RxQueue(args.get_arg('rx_if')) + txq = TxQueue(args.get_arg('tx_if')) + + src_mac = args.get_arg('src_mac') + dst_mac = args.get_arg('dst_mac') + src_ip = args.get_arg('src_ip') + dst_ip = args.get_arg('dst_ip') + start_size = int(args.get_arg('start_size')) + end_size = int(args.get_arg('end_size')) + step = int(args.get_arg('step')) + echo_id = 0xa + # generate some random data buffer + data = bytearray(os.urandom(end_size)) + + # send ICMPv6 neighbor advertisement message + sent_packets = [] + pkt_send = (Ether(src=src_mac, dst=dst_mac) / + IPv6(src=src_ip, dst=dst_ip) / + ICMPv6ND_NA(tgt=src_ip, R=0) / + ICMPv6NDOptDstLLAddr(lladdr=src_mac)) + sent_packets.append(pkt_send) + txq.send(pkt_send) + + # send ICMPv6 echo request with incremented data length and receive ICMPv6 + # echo reply + for echo_seq in range(start_size, end_size, step): + pkt_send = (Ether(src=src_mac, dst=dst_mac) / + IPv6(src=src_ip, dst=dst_ip) / + ICMPv6EchoRequest(id=echo_id, seq=echo_seq, + data=data[0:echo_seq])) + sent_packets.append(pkt_send) + txq.send(pkt_send) + + ether = rxq.recv(ignore=sent_packets) + if ether is None: + rxq._proc.terminate() + raise RuntimeError( + 'ICMPv6 echo reply seq {0} Rx timeout'.format(echo_seq)) + + if not ether.haslayer(IPv6): + rxq._proc.terminate() + raise RuntimeError( + 'Unexpected packet with no IPv6 received {0}'.format( + ether.__repr__())) + + ipv6 = ether['IPv6'] + + if not ipv6.haslayer(ICMPv6EchoReply): + rxq._proc.terminate() + raise RuntimeError( + 'Unexpected packet with no IPv6 ICMP received {0}'.format( + ipv6.__repr__())) + + icmpv6 = ipv6['ICMPv6 Echo Reply'] + + if icmpv6.id != echo_id or icmpv6.seq != echo_seq: + rxq._proc.terminate() + raise RuntimeError( + 'Invalid ICMPv6 echo reply received ID {0} seq {1} should be ' + + 'ID {2} seq {3}, {0}'.format(icmpv6.id, icmpv6.seq, echo_id, + echo_seq)) + + cksum = icmpv6.cksum + del icmpv6.cksum + tmp = ICMPv6EchoReply(str(icmpv6)) + if tmp.cksum != cksum: + rxq._proc.terminate() + raise RuntimeError( + 'Invalid checksum {0} should be {1}'.format(cksum, tmp.cksum)) + + rxq._proc.terminate() + sys.exit(0) + +if __name__ == "__main__": + main() diff --git a/resources/traffic_scripts/send_ip_icmp.py b/resources/traffic_scripts/send_ip_icmp.py new file mode 100755 index 0000000000..fd15376fb1 --- /dev/null +++ b/resources/traffic_scripts/send_ip_icmp.py @@ -0,0 +1,71 @@ +#!/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 ip icmp packet +from one interface to the other""" + +import sys +from resources.libraries.python.PacketVerifier import RxQueue, TxQueue +from resources.libraries.python.TrafficScriptArg import TrafficScriptArg +from scapy.layers.inet import ICMP, IP +from scapy.all import Ether + + +def main(): + """ Send IP icmp packet from one traffic generator interface to the other""" + args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip']) + + src_mac = args.get_arg('src_mac') + dst_mac = args.get_arg('dst_mac') + src_ip = args.get_arg('src_ip') + dst_ip = args.get_arg('dst_ip') + tx_if = args.get_arg('tx_if') + rx_if = args.get_arg('rx_if') + + rxq = RxQueue(rx_if) + txq = TxQueue(tx_if) + + sent_packets = [] + + # Create empty ip ICMP packet and add padding before sending + pkt_raw = Ether(src=src_mac, dst=dst_mac) / \ + IP(src=src_ip, dst=dst_ip) / \ + ICMP() + + # Send created packet on one interface and receive on the other + sent_packets.append(pkt_raw) + txq.send(pkt_raw) + + ether = rxq.recv(1) + + # Check whether received packet contains layers Ether, IP and ICMP + if ether is None: + rxq._proc.terminate() + raise RuntimeError('ICMPv6 echo reply Rx timeout') + + if not ether.haslayer(IP): + rxq._proc.terminate() + raise RuntimeError( + 'Not an IP packet received {0}'.format(ether.__repr__())) + + if not ether.haslayer(ICMP): + rxq._proc.terminate() + raise RuntimeError( + 'Not an ICMP packet received {0}'.format(ether.__repr__())) + + rxq._proc.terminate() + sys.exit(0) + +if __name__ == "__main__": + main() |