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
  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         elif ether.haslayer(ICMPv6MLReport2):
 129             # read another packet in the queue if the current one is
 130             # ICMPv6MLReport2
 131             continue
 132         else:
 133             # otherwise process the current packet
 134             break
 135
 136     if rx_dst_mac == ether[Ether].dst and rx_src_mac == ether[Ether].src:
 137         logger.trace(u"MAC matched")
 138     else:
 139         raise RuntimeError(f"Matching packet unsuccessful: {ether!r}")
 140
 141     if encaps_rx == u"Dot1q":
 142         if ether[Dot1Q].vlan == int(vlan_rx):
 143             logger.trace(u"VLAN matched")
 144         else:
 145             raise RuntimeError(
 146                 f"Ethernet frame with wrong VLAN tag "
 147                 f"({ether[Dot1Q].vlan}-received, "
 148                 f"{vlan_rx}-expected):\n{ether!r}"
 149             )
 150         ip = ether[Dot1Q].payload
 151     elif encaps_rx == u"Dot1ad":
 152         raise NotImplementedError()
 153     else:
 154         ip = ether.payload
 155
 156     if not isinstance(ip, ip_format):
 157         raise RuntimeError(f"Not an IP packet received {ip!r}")
 158
 159     # Compare data from packets
 160     if src_ip == ip.src:
 161         logger.trace(u"Src IP matched")
 162     else:
 163         raise RuntimeError(
 164             f"Matching Src IP unsuccessful: {src_ip} != {ip.src}"
 165         )
 166
 167     if dst_ip == ip.dst:
 168         logger.trace(u"Dst IP matched")
 169     else:
 170         raise RuntimeError(
 171             f"Matching Dst IP unsuccessful: {dst_ip} != {ip.dst}"
 172         )
 173
 174     sys.exit(0)
 175
 176
 177 if __name__ == u"__main__":
 178     main()