GPL/traffic_scripts/send_ip_check_headers.py

   1 #!/usr/bin/env python3
   2
   3 # Copyright (c) 2021 Cisco and/or its affiliates.
   4 #
   5 # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
   6 #
   7 # Licensed under the Apache License 2.0 or
   8 # GNU General Public License v2.0 or later;  you may not use this file
   9 # except in compliance with one of these Licenses. You
  10 # may obtain a copy of the Licenses at:
  11 #
  12 #     http://www.apache.org/licenses/LICENSE-2.0
  13 #     https://www.gnu.org/licenses/old-licenses/gpl-2.0-standalone.html
  14 #
  15 # Note: If this file is linked with Scapy, which is GPLv2+, your use of it
  16 # must be under GPLv2+.  If at any point in the future it is no longer linked
  17 # with Scapy (or other GPLv2+ licensed software), you are free to choose
  18 # Apache 2.
  19 #
  20 # Unless required by applicable law or agreed to in writing, software
  21 # distributed under the License is distributed on an "AS IS" BASIS,
  22 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  23 # See the License for the specific language governing permissions and
  24 # limitations under the License.
  25
  26 """Traffic script that sends an IP IPv4/IPv6 packet from one interface
  27 to the other. Source and destination IP addresses and source and destination
  28 MAC addresses are checked in received packet.
  29 """
  30
  31 import sys
  32
  33 import ipaddress
  34
  35 from robot.api import logger
  36 from scapy.layers.inet import IP
  37 from scapy.layers.inet6 import IPv6, ICMPv6ND_NS
  38 from scapy.layers.l2 import Ether, Dot1Q
  39 from scapy.packet import Raw
  40
  41 from .PacketVerifier import RxQueue, TxQueue
  42 from .TrafficScriptArg import TrafficScriptArg
  43
  44
  45 def valid_ipv4(ip):
  46     try:
  47         ipaddress.IPv4Address(ip)
  48         return True
  49     except (AttributeError, ipaddress.AddressValueError):
  50         return False
  51
  52
  53 def valid_ipv6(ip):
  54     try:
  55         ipaddress.IPv6Address(ip)
  56         return True
  57     except (AttributeError, ipaddress.AddressValueError):
  58         return False
  59
  60
  61 def main():
  62     """Send IP/IPv6 packet from one traffic generator interface to the other."""
  63     args = TrafficScriptArg(
  64         [
  65             u"tg_src_mac", u"tg_dst_mac", u"src_ip", u"dst_ip", u"dut_if1_mac",
  66             u"dut_if2_mac"
  67         ],
  68         [
  69             u"encaps_tx", u"vlan_tx", u"vlan_outer_tx", u"encaps_rx",
  70             u"vlan_rx", u"vlan_outer_rx"
  71         ]
  72     )
  73
  74     tx_src_mac = args.get_arg(u"tg_src_mac")
  75     tx_dst_mac = args.get_arg(u"dut_if1_mac")
  76     rx_dst_mac = args.get_arg(u"tg_dst_mac")
  77     rx_src_mac = args.get_arg(u"dut_if2_mac")
  78     src_ip = args.get_arg(u"src_ip")
  79     dst_ip = args.get_arg(u"dst_ip")
  80     tx_if = args.get_arg(u"tx_if")
  81     rx_if = args.get_arg(u"rx_if")
  82
  83     encaps_tx = args.get_arg(u"encaps_tx")
  84     vlan_tx = args.get_arg(u"vlan_tx")
  85     vlan_outer_tx = args.get_arg(u"vlan_outer_tx")
  86     encaps_rx = args.get_arg(u"encaps_rx")
  87     vlan_rx = args.get_arg(u"vlan_rx")
  88     vlan_outer_rx = args.get_arg(u"vlan_outer_rx")
  89
  90     rxq = RxQueue(rx_if)
  91     txq = TxQueue(tx_if)
  92
  93     sent_packets =list()
  94     pkt_raw = Ether(src=tx_src_mac, dst=tx_dst_mac)
  95
  96     if encaps_tx == u"Dot1q":
  97         pkt_raw /= Dot1Q(vlan=int(vlan_tx))
  98     elif encaps_tx == u"Dot1ad":
  99         pkt_raw.type = 0x88a8
 100         pkt_raw /= Dot1Q(vlan=vlan_outer_tx)
 101         pkt_raw /= Dot1Q(vlan=vlan_tx)
 102
 103     if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
 104         pkt_raw /= IP(src=src_ip, dst=dst_ip, proto=61)
 105         ip_format = IP
 106     elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
 107         pkt_raw /= IPv6(src=src_ip, dst=dst_ip)
 108         ip_format = IPv6
 109     else:
 110         raise ValueError(u"IP not in correct format")
 111
 112     pkt_raw /= Raw()
 113     sent_packets.append(pkt_raw)
 114     txq.send(pkt_raw)
 115
 116     while True:
 117         if tx_if == rx_if:
 118             ether = rxq.recv(2, ignore=sent_packets)
 119         else:
 120             ether = rxq.recv(2)
 121
 122         if ether is None:
 123             raise RuntimeError(u"IP packet Rx timeout")
 124
 125         if ether.haslayer(ICMPv6ND_NS):
 126             # read another packet in the queue if the current one is ICMPv6ND_NS
 127             continue
 128         else:
 129             # otherwise process the current packet
 130             break
 131
 132     if rx_dst_mac == ether[Ether].dst and rx_src_mac == ether[Ether].src:
 133         logger.trace(u"MAC matched")
 134     else:
 135         raise RuntimeError(f"Matching packet unsuccessful: {ether!r}")
 136
 137     if encaps_rx == u"Dot1q":
 138         if ether[Dot1Q].vlan == int(vlan_rx):
 139             logger.trace(u"VLAN matched")
 140         else:
 141             raise RuntimeError(
 142                 f"Ethernet frame with wrong VLAN tag "
 143                 f"({ether[Dot1Q].vlan}-received, "
 144                 f"{vlan_rx}-expected):\n{ether!r}"
 145             )
 146         ip = ether[Dot1Q].payload
 147     elif encaps_rx == u"Dot1ad":
 148         raise NotImplementedError()
 149     else:
 150         ip = ether.payload
 151
 152     if not isinstance(ip, ip_format):
 153         raise RuntimeError(f"Not an IP packet received {ip!r}")
 154
 155     # Compare data from packets
 156     if src_ip == ip.src:
 157         logger.trace(u"Src IP matched")
 158     else:
 159         raise RuntimeError(
 160             f"Matching Src IP unsuccessful: {src_ip} != {ip.src}"
 161         )
 162
 163     if dst_ip == ip.dst:
 164         logger.trace(u"Dst IP matched")
 165     else:
 166         raise RuntimeError(
 167             f"Matching Dst IP unsuccessful: {dst_ip} != {ip.dst}"
 168         )
 169
 170     sys.exit(0)
 171
 172
 173 if __name__ == u"__main__":
 174     main()