X-Git-Url: https://gerrit.fd.io/r/gitweb?p=csit.git;a=blobdiff_plain;f=resources%2Ftraffic_scripts%2Fsend_ip_icmp.py;h=58f2e1e4d8d10d67b0f9e6a5b4d37ec6d2f08ef1;hp=5e365ebc38252f390632d6d1c3e0d6977e52dbcd;hb=5ad9b364cbd45a0b25d73412b9777ac14df92b0a;hpb=84c45de446b69f83df919a91bc684da910824d6f diff --git a/resources/traffic_scripts/send_ip_icmp.py b/resources/traffic_scripts/send_ip_icmp.py index 5e365ebc38..58f2e1e4d8 100755 --- a/resources/traffic_scripts/send_ip_icmp.py +++ b/resources/traffic_scripts/send_ip_icmp.py @@ -12,24 +12,73 @@ # 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""" +"""Traffic script that sends an IP ICMPv4/ICMPv6 packet from one interface to +the other one. Dot1q or Dot1ad tagging of the ethernet frame can be set. +""" import sys +import ipaddress + +from scapy.layers.l2 import Ether, Dot1Q +from scapy.layers.inet import ICMP, IP +from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6ND_NS + 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 valid_ipv4(ip): + """Check if IP address has the correct IPv4 address format. + + :param ip: IP address. + :type ip: str + :return: True in case of correct IPv4 address format, + otherwise return false. + :rtype: bool + """ + try: + ipaddress.IPv4Address(unicode(ip)) + return True + except (AttributeError, ipaddress.AddressValueError): + return False + + +def valid_ipv6(ip): + """Check if IP address has the correct IPv6 address format. + + :param ip: IP address. + :type ip: str + :return: True in case of correct IPv6 address format, + otherwise return false. + :rtype: bool + """ + try: + ipaddress.IPv6Address(unicode(ip)) + return True + except (AttributeError, ipaddress.AddressValueError): + return False def main(): - """ Send IP icmp packet from one traffic generator interface to the other""" - args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip']) + """Send IP ICMPv4/ICMPv6 packet from one traffic generator interface to + the other one. Dot1q or Dot1ad tagging of the ethernet frame can be set. + """ + args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip'], + ['encaps', 'vlan1', 'vlan2', 'encaps_rx', + 'vlan1_rx', 'vlan2_rx']) 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') + + encaps = args.get_arg('encaps') + vlan1 = args.get_arg('vlan1') + vlan2 = args.get_arg('vlan2') + encaps_rx = args.get_arg('encaps_rx') + vlan1_rx = args.get_arg('vlan1_rx') + vlan2_rx = args.get_arg('vlan2_rx') + tx_if = args.get_arg('tx_if') rx_if = args.get_arg('rx_if') @@ -37,31 +86,101 @@ def main(): txq = TxQueue(tx_if) sent_packets = [] - + ip_format = '' + icmp_format = '' # 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() + if valid_ipv4(src_ip) and valid_ipv4(dst_ip): + if encaps == 'Dot1q': + pkt_raw = (Ether(src=src_mac, dst=dst_mac) / + Dot1Q(vlan=int(vlan1)) / + IP(src=src_ip, dst=dst_ip) / + ICMP()) + elif encaps == 'Dot1ad': + pkt_raw = (Ether(src=src_mac, dst=dst_mac, type=0x88A8) / + Dot1Q(vlan=int(vlan1), type=0x8100) / + Dot1Q(vlan=int(vlan2)) / + IP(src=src_ip, dst=dst_ip) / + ICMP()) + else: + pkt_raw = (Ether(src=src_mac, dst=dst_mac) / + IP(src=src_ip, dst=dst_ip) / + ICMP()) + ip_format = IP + icmp_format = ICMP + elif valid_ipv6(src_ip) and valid_ipv6(dst_ip): + if encaps == 'Dot1q': + pkt_raw = (Ether(src=src_mac, dst=dst_mac) / + Dot1Q(vlan=int(vlan1)) / + IPv6(src=src_ip, dst=dst_ip) / + ICMPv6EchoRequest()) + elif encaps == 'Dot1ad': + pkt_raw = (Ether(src=src_mac, dst=dst_mac, type=0x88A8) / + Dot1Q(vlan=int(vlan1), type=0x8100) / + Dot1Q(vlan=int(vlan2)) / + IPv6(src=src_ip, dst=dst_ip) / + ICMPv6EchoRequest()) + else: + pkt_raw = (Ether(src=src_mac, dst=dst_mac) / + IPv6(src=src_ip, dst=dst_ip) / + ICMPv6EchoRequest()) + ip_format = IPv6 + icmp_format = ICMPv6EchoRequest + else: + raise ValueError("IP(s) not in correct format") # Send created packet on one interface and receive on the other sent_packets.append(pkt_raw) txq.send(pkt_raw) - ether = rxq.recv(10) + # Receive ICMP / ICMPv6 echo reply + while True: + ether = rxq.recv(2,) + if ether is None: + raise RuntimeError('ICMP echo Rx timeout') + + if ether.haslayer(ICMPv6ND_NS): + # read another packet in the queue if the current one is ICMPv6ND_NS + continue + else: + # otherwise process the current packet + break - # Check whether received packet contains layers Ether, IP and ICMP - if ether is None: - raise RuntimeError('ICMPv6 echo reply Rx timeout') + # Check whether received packet contains layers IP/IPv6 and + # ICMP/ICMPv6EchoRequest + if encaps_rx: + if encaps_rx == 'Dot1q': + if not vlan1_rx: + vlan1_rx = vlan1 + if not ether.haslayer(Dot1Q): + raise RuntimeError('Not VLAN tagged Eth frame received:\n{0}'. + format(ether.__repr__())) + elif ether[Dot1Q].vlan != int(vlan1_rx): + raise RuntimeError('Ethernet frame with wrong VLAN tag ({}) ' + 'received ({} expected):\n{}'. + format(ether[Dot1Q].vlan, vlan1_rx, + ether.__repr__())) + elif encaps_rx == 'Dot1ad': + if not vlan1_rx: + vlan1_rx = vlan1 + if not vlan2_rx: + vlan2_rx = vlan2 + # TODO + raise RuntimeError('Encapsulation {0} not implemented yet.'. + format(encaps_rx)) + else: + raise RuntimeError('Unsupported encapsulation expected: {0}'. + format(encaps_rx)) - if not ether.haslayer(IP): - raise RuntimeError( - 'Not an IP packet received {0}'.format(ether.__repr__())) + if not ether.haslayer(ip_format): + raise RuntimeError('Not an IP/IPv6 packet received:\n{0}'. + format(ether.__repr__())) - if not ether.haslayer(ICMP): - raise RuntimeError( - 'Not an ICMP packet received {0}'.format(ether.__repr__())) + if not ether.haslayer(icmp_format): + raise RuntimeError('Not an ICMP/ICMPv6EchoRequest packet received:\n' + '{0}'.format(ether.__repr__())) sys.exit(0) + if __name__ == "__main__": main()