resources/traffic_scripts/send_ipv4_check_lw_4o6.py

   1 #!/usr/bin/env python
   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:
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 """Traffic script that sends an empty UDP datagram and checks if IPv4 is
  16 correctly encapsulated into IPv6 packet."""
  17
  18 import sys
  19
  20 from scapy.layers.l2 import Ether
  21 from scapy.layers.inet import IP, UDP
  22
  23 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
  24 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  25
  26
  27 def _is_ipv4_in_ipv6(pkt):
  28     """If IPv6 next header type in the given pkt is IPv4, return True,
  29     else return False. False is returned also if exception occurs."""
  30     ipv6_type = int('0x86dd', 16)  # IPv6
  31     try:
  32         if pkt.type == ipv6_type:
  33             if pkt.payload.nh == 4:
  34                 return True
  35     except:  # pylint: disable=bare-except
  36         return False
  37     return False
  38
  39
  40 def main():  # pylint: disable=too-many-statements, too-many-locals
  41     """Main function of the script file."""
  42     args = TrafficScriptArg(['tx_dst_mac', 'tx_src_ipv4', 'tx_dst_ipv4',
  43                              'tx_dst_udp_port', 'rx_dst_mac', 'rx_src_mac',
  44                              'src_ipv6', 'dst_ipv6'])
  45     rx_if = args.get_arg('rx_if')
  46     tx_if = args.get_arg('tx_if')
  47     tx_dst_mac = args.get_arg('tx_dst_mac')
  48     tx_src_ipv4 = args.get_arg('tx_src_ipv4')
  49     tx_dst_ipv4 = args.get_arg('tx_dst_ipv4')
  50     tx_dst_udp_port = int(args.get_arg('tx_dst_udp_port'))
  51     tx_src_udp_port = 20000
  52     rx_dst_mac = args.get_arg('rx_dst_mac')
  53     rx_src_mac = args.get_arg('rx_src_mac')
  54     rx_src_ipv6 = args.get_arg('src_ipv6')
  55     rx_dst_ipv6 = args.get_arg('dst_ipv6')
  56
  57     rxq = RxQueue(rx_if)
  58     txq = TxQueue(tx_if)
  59     sent_packets = []
  60
  61     # Create empty UDP datagram
  62     udp = (Ether(dst=tx_dst_mac) /
  63            IP(src=tx_src_ipv4, dst=tx_dst_ipv4) /
  64            UDP(sport=tx_src_udp_port, dport=tx_dst_udp_port))
  65
  66     txq.send(udp)
  67     sent_packets.append(udp)
  68
  69     for _ in range(5):
  70         pkt = rxq.recv(2)
  71         if _is_ipv4_in_ipv6(pkt):
  72             ether = pkt
  73             break
  74     else:
  75         raise RuntimeError("IPv4 in IPv6 Rx error.")
  76
  77     # check ethernet
  78     if ether.dst != rx_dst_mac:
  79         raise RuntimeError("Destination MAC error {} != {}.".
  80                            format(ether.dst, rx_dst_mac))
  81     print "Destination MAC: OK."
  82
  83     if ether.src != rx_src_mac:
  84         raise RuntimeError("Source MAC error {} != {}.".
  85                            format(ether.src, rx_src_mac))
  86     print "Source MAC: OK."
  87
  88     ipv6 = ether.payload
  89
  90     # check ipv6
  91     if ipv6.dst != rx_dst_ipv6:
  92         raise RuntimeError("Destination IP error {} != {}.".
  93                            format(ipv6.dst, rx_dst_ipv6))
  94     print "Destination IPv6: OK."
  95
  96     if ipv6.src != rx_src_ipv6:
  97         raise RuntimeError("Source IP error {} != {}.".
  98                            format(ipv6.src, rx_src_ipv6))
  99     print "Source IPv6: OK."
 100
 101     ipv4 = ipv6.payload
 102
 103     # check ipv4
 104     if ipv4.dst != tx_dst_ipv4:
 105         raise RuntimeError("Destination IP error {} != {}.".
 106                            format(ipv4.dst, tx_dst_ipv4))
 107     print "Destination IPv4: OK."
 108
 109     if ipv4.src != tx_src_ipv4:
 110         raise RuntimeError("Source IP error {} != {}.".
 111                            format(ipv4.src, tx_src_ipv4))
 112     print "Source IPv4: OK."
 113
 114     if ipv4.proto != 17:  # UDP
 115         raise RuntimeError("IP protocol error {} != UDP.".
 116                            format(ipv4.proto))
 117     print "IPv4 protocol: OK."
 118
 119     udp = ipv4.payload
 120
 121     # check udp
 122     if udp.dport != tx_dst_udp_port:
 123         raise RuntimeError("UDP dport error {} != {}.".
 124                            format(udp.dport, tx_dst_udp_port))
 125     print "UDP dport: OK."
 126
 127     if udp.sport != tx_src_udp_port:
 128         raise RuntimeError("UDP sport error {} != {}.".
 129                            format(udp.sport, tx_src_udp_port))
 130     print "UDP sport: OK."
 131
 132     sys.exit(0)
 133
 134 if __name__ == "__main__":
 135     main()