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, ICMPv6MLReport2, ICMPv6ND_RA
  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_address):
  46     """Check IPv4 address.
  47
  48     :param ip_address: IPv4 address to check.
  49     :type ip_address: str
  50     :returns: True if IP address is correct.
  51     :rtype: bool
  52     """
  53     try:
  54         ipaddress.IPv4Address(ip_address)
  55         return True
  56     except (AttributeError, ipaddress.AddressValueError):
  57         return False
  58
  59
  60 def valid_ipv6(ip_address):
  61     """Check IPv6 address.
  62
  63     :param ip_address: IPv6 address to check.
  64     :type ip_address: str
  65     :returns: True if IP address is correct.
  66     :rtype: bool
  67     """
  68     try:
  69         ipaddress.IPv6Address(ip_address)
  70         return True
  71     except (AttributeError, ipaddress.AddressValueError):
  72         return False
  73
  74
  75 def main():
  76     """Send IP/IPv6 packet from one traffic generator interface to the other."""
  77     args = TrafficScriptArg(
  78         [
  79             u"tg_src_mac", u"tg_dst_mac", u"src_ip", u"dst_ip", u"dut_if1_mac",
  80             u"dut_if2_mac"
  81         ],
  82         [
  83             u"encaps_tx", u"vlan_tx", u"vlan_outer_tx", u"encaps_rx",
  84             u"vlan_rx", u"vlan_outer_rx"
  85         ]
  86     )
  87
  88     tx_src_mac = args.get_arg(u"tg_src_mac")
  89     tx_dst_mac = args.get_arg(u"dut_if1_mac")
  90     rx_dst_mac = args.get_arg(u"tg_dst_mac")
  91     rx_src_mac = args.get_arg(u"dut_if2_mac")
  92     src_ip = args.get_arg(u"src_ip")
  93     dst_ip = args.get_arg(u"dst_ip")
  94     tx_if = args.get_arg(u"tx_if")
  95     rx_if = args.get_arg(u"rx_if")
  96
  97     encaps_tx = args.get_arg(u"encaps_tx")
  98     vlan_tx = args.get_arg(u"vlan_tx")
  99     vlan_outer_tx = args.get_arg(u"vlan_outer_tx")
 100     encaps_rx = args.get_arg(u"encaps_rx")
 101     vlan_rx = args.get_arg(u"vlan_rx")
 102     vlan_outer_rx = args.get_arg(u"vlan_outer_rx")
 103
 104     rxq = RxQueue(rx_if)
 105     txq = TxQueue(tx_if)
 106
 107     sent_packets = list()
 108     pkt_raw = Ether(src=tx_src_mac, dst=tx_dst_mac)
 109
 110     if encaps_tx == u"Dot1q":
 111         pkt_raw /= Dot1Q(vlan=int(vlan_tx))
 112     elif encaps_tx == u"Dot1ad":
 113         pkt_raw.type = 0x88a8
 114         pkt_raw /= Dot1Q(vlan=vlan_outer_tx)
 115         pkt_raw /= Dot1Q(vlan=vlan_tx)
 116
 117     if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
 118         pkt_raw /= IP(src=src_ip, dst=dst_ip, proto=61)
 119         ip_format = IP
 120     elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
 121         pkt_raw /= IPv6(src=src_ip, dst=dst_ip)
 122         ip_format = IPv6
 123     else:
 124         raise ValueError(u"IP not in correct format")
 125
 126     pkt_raw /= Raw()
 127     sent_packets.append(pkt_raw)
 128     txq.send(pkt_raw)
 129
 130     while True:
 131         if tx_if == rx_if:
 132             ether = rxq.recv(2, ignore=sent_packets)
 133         else:
 134             ether = rxq.recv(2)
 135
 136         if ether is None:
 137             raise RuntimeError(u"IP packet Rx timeout")
 138
 139         if ether.haslayer(ICMPv6ND_NS):
 140             # read another packet in the queue if the current one is ICMPv6ND_NS
 141             continue
 142         elif ether.haslayer(ICMPv6MLReport2):
 143             # read another packet in the queue if the current one is
 144             # ICMPv6MLReport2
 145             continue
 146         elif ether.haslayer(ICMPv6ND_RA):
 147             # read another packet in the queue if the current one is
 148             # ICMPv6ND_RA
 149             continue
 150
 151         break
 152
 153     if rx_dst_mac == ether[Ether].dst and rx_src_mac == ether[Ether].src:
 154         logger.trace(u"MAC matched")
 155     else:
 156         raise RuntimeError(f"Matching packet unsuccessful: {ether!r}")
 157
 158     if encaps_rx == u"Dot1q":
 159         if ether[Dot1Q].vlan == int(vlan_rx):
 160             logger.trace(u"VLAN matched")
 161         else:
 162             raise RuntimeError(
 163                 f"Ethernet frame with wrong VLAN tag "
 164                 f"({ether[Dot1Q].vlan}-received, "
 165                 f"{vlan_rx}-expected):\n{ether!r}"
 166             )
 167         ip = ether[Dot1Q].payload
 168     elif encaps_rx == u"Dot1ad":
 169         raise NotImplementedError()
 170     else:
 171         ip = ether.payload
 172
 173     if not isinstance(ip, ip_format):
 174         raise RuntimeError(f"Not an IP packet received {ip!r}")
 175
 176     # Compare data from packets
 177     if src_ip == ip.src:
 178         logger.trace(u"Src IP matched")
 179     else:
 180         raise RuntimeError(
 181             f"Matching Src IP unsuccessful: {src_ip} != {ip.src}"
 182         )
 183
 184     if dst_ip == ip.dst:
 185         logger.trace(u"Dst IP matched")
 186     else:
 187         raise RuntimeError(
 188             f"Matching Dst IP unsuccessful: {dst_ip} != {ip.dst}"
 189         )
 190
 191     sys.exit(0)
 192
 193
 194 if __name__ == u"__main__":
 195     main()