2 # Copyright (c) 2016 Cisco and/or its affiliates.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at:
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
15 """Traffic script that sends an IP ICMPv4/ICMPv6 packet from one interface
16 to the other. Source and destination IP addresses and source and destination
17 MAC addresses are checked in received packet.
23 from scapy.layers.inet import IP, ICMP, ARP
24 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
25 from scapy.layers.l2 import Ether
27 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue, auto_pad
28 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
31 def valid_ipv4(address):
32 """Check if IP address has the correct IPv4 address format.
34 :param address: IP address.
36 :return: True in case of correct IPv4 address format,
37 otherwise return false.
41 ipaddress.IPv4Address(unicode(address))
43 except (AttributeError, ipaddress.AddressValueError):
47 def valid_ipv6(address):
48 """Check if IP address has the correct IPv6 address format.
50 :param address: IP address.
52 :return: True in case of correct IPv6 address format,
53 otherwise return false.
57 ipaddress.IPv6Address(unicode(address))
59 except (AttributeError, ipaddress.AddressValueError):
64 """Send a simple L2 or ICMP packet from one TG interface to DUT, then
65 receive a copy of the packet on the second TG interface, and a copy of
67 args = TrafficScriptArg(
68 ['tg_src_mac', 'src_ip', 'dst_ip', 'dut_if1_mac', 'ptype'])
70 src_mac = args.get_arg('tg_src_mac')
71 dst_mac = args.get_arg('dut_if1_mac')
72 src_ip = args.get_arg('src_ip')
73 dst_ip = args.get_arg('dst_ip')
74 tx_if = args.get_arg('tx_if')
75 rx_if = args.get_arg('rx_if')
76 ptype = args.get_arg('ptype')
82 pkt_raw = (Ether(src=src_mac, dst=dst_mac) /
83 ARP(hwsrc=src_mac, hwdst="00:00:00:00:00:00",
84 psrc=src_ip, pdst=dst_ip, op="who-has"))
86 if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
87 pkt_raw = (Ether(src=src_mac, dst=dst_mac) /
88 IP(src=src_ip, dst=dst_ip) /
89 ICMP(type="echo-request"))
90 elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
91 pkt_raw = (Ether(src=src_mac, dst=dst_mac) /
92 IPv6(src=src_ip, dst=dst_ip) /
95 raise ValueError("IP not in correct format")
97 raise RuntimeError("Unexpected payload type.")
102 # Receive copy of sent packet.
104 raise RuntimeError("Rx timeout")
105 pkt = auto_pad(pkt_raw)
106 if str(ether) != str(pkt):
107 raise RuntimeError("Mirrored packet does not match packet sent.")
109 # Receive copy of reply to sent packet.
112 raise RuntimeError("Rx timeout")
113 if ether.src != dst_mac or ether.dst != src_mac:
114 raise RuntimeError("MAC mismatch in mirrored response.")
116 if ether['ARP'].op != 2:
117 raise RuntimeError("Mirrored packet is not an ARP reply.")
118 if ether['ARP'].hwsrc != dst_mac or ether['ARP'].hwdst != src_mac:
119 raise RuntimeError("ARP MAC does not match l2 MAC "
120 "in mirrored response.")
121 if ether['ARP'].psrc != dst_ip or ether['ARP'].pdst != src_ip:
122 raise RuntimeError("ARP IP address mismatch in mirrored response.")
123 elif ptype == "ICMP" and ether.haslayer(IP):
124 if ether['IP'].src != dst_ip or ether['IP'].dst != src_ip:
125 raise RuntimeError("IP address mismatch in mirrored reply.")
126 if ether['ICMP'].type != 0:
127 raise RuntimeError("Mirrored packet is not an ICMP reply.")
128 elif ptype == "ICMP" and ether.haslayer(IPv6):
129 if ether['IPv6'].src != dst_ip or ether['IPv6'].dst != src_ip:
130 raise RuntimeError("IP address mismatch in mirrored reply.")
131 if not ether.haslayer(ICMPv6EchoReply):
132 raise RuntimeError("Mirrored packet is not an ICMP reply.")
137 if __name__ == "__main__":