GPL/traffic_scripts/send_vxlan_check_vxlan.py

   1 #!/usr/bin/env python3
   2
   3 # Copyright (c) 2020 Cisco and/or its affiliates.
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at:
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 """Traffic script that sends an IP ICMPv4/ICMPv6 packet from one interface to
  17 the other one. Dot1q or Dot1ad tagging of the ethernet frame can be set.
  18 """
  19
  20 import sys
  21
  22 from scapy.layers.inet import IP, UDP
  23 from scapy.layers.l2 import Ether
  24 from scapy.packet import Raw
  25
  26 from .PacketVerifier import RxQueue, TxQueue
  27 from .TrafficScriptArg import TrafficScriptArg
  28 from . import vxlan
  29
  30
  31 def main():
  32     """Send IP ICMPv4/ICMPv6 packet from one traffic generator interface to
  33     the other one. Dot1q or Dot1ad tagging of the ethernet frame can be set.
  34     """
  35     args = TrafficScriptArg(
  36         [
  37             u"tx_src_mac", u"tx_dst_mac", u"tx_src_ip", u"tx_dst_ip", u"tx_vni",
  38             u"rx_src_ip", u"rx_dst_ip", u"rx_vni"
  39         ]
  40     )
  41
  42     tx_if = args.get_arg(u"tx_if")
  43     rx_if = args.get_arg(u"rx_if")
  44     tx_src_mac = args.get_arg(u"tx_src_mac")
  45     tx_dst_mac = args.get_arg(u"tx_dst_mac")
  46     tx_src_ip = args.get_arg(u"tx_src_ip")
  47     tx_dst_ip = args.get_arg(u"tx_dst_ip")
  48     tx_vni = args.get_arg(u"tx_vni")
  49     rx_src_ip = args.get_arg(u"rx_src_ip")
  50     rx_dst_ip = args.get_arg(u"rx_dst_ip")
  51     rx_vni = args.get_arg(u"rx_vni")
  52
  53     rxq = RxQueue(rx_if)
  54     txq = TxQueue(tx_if)
  55
  56     sent_packets = []
  57
  58     tx_pkt_p = (Ether(src=u"02:00:00:00:00:01", dst=u"02:00:00:00:00:02") /
  59                 IP(src=u"192.168.1.1", dst=u"192.168.1.2") /
  60                 UDP(sport=12345, dport=1234) /
  61                 Raw(u"raw data"))
  62
  63     pkt_raw = (Ether(src=tx_src_mac, dst=tx_dst_mac) /
  64                IP(src=tx_src_ip, dst=tx_dst_ip) /
  65                UDP(sport=23456) /
  66                vxlan.VXLAN(vni=int(tx_vni)) /
  67                tx_pkt_p)
  68
  69     pkt_raw /= Raw()
  70     # Send created packet on one interface and receive on the other
  71     sent_packets.append(pkt_raw)
  72     txq.send(pkt_raw)
  73
  74     ether = rxq.recv(2, ignore=sent_packets)
  75
  76     # Check whether received packet contains layers Ether, IP and VXLAN
  77     if ether is None:
  78         raise RuntimeError(u"Packet Rx timeout")
  79     ip = ether.payload
  80
  81     if ip.src != rx_src_ip:
  82         raise RuntimeError(f"IP src mismatch {ip.src} != {rx_src_ip}")
  83     if ip.dst != rx_dst_ip:
  84         raise RuntimeError(f"IP dst mismatch {ip.dst} != {rx_dst_ip}")
  85     if ip.payload.dport != 4789:
  86         raise RuntimeError(
  87             f"VXLAN UDP port mismatch {ip.payload.dport} != 4789"
  88         )
  89     vxlan_pkt = ip.payload.payload
  90
  91     if int(vxlan_pkt.vni) != int(rx_vni):
  92         raise RuntimeError(u"vxlan mismatch")
  93     rx_pkt_p = vxlan_pkt.payload
  94
  95     if rx_pkt_p.src != tx_pkt_p.src:
  96         raise RuntimeError(
  97             f"RX encapsulated MAC src mismatch {rx_pkt_p.src} != {tx_pkt_p.src}"
  98         )
  99     if rx_pkt_p.dst != tx_pkt_p.dst:
 100         raise RuntimeError(
 101             f"RX encapsulated MAC dst mismatch {rx_pkt_p.dst} != {tx_pkt_p.dst}"
 102         )
 103     if rx_pkt_p[IP].src != tx_pkt_p[IP].src:
 104         raise RuntimeError(
 105             f"RX encapsulated IP src mismatch {rx_pkt_p[IP].src} != "
 106             f"{tx_pkt_p[IP].src}"
 107         )
 108     if rx_pkt_p[IP].dst != tx_pkt_p[IP].dst:
 109         raise RuntimeError(
 110             f"RX encapsulated IP dst mismatch {rx_pkt_p[IP].dst} != "
 111             f"{tx_pkt_p[IP].dst}"
 112         )
 113
 114     # TODO: verify inner Ether()
 115
 116     sys.exit(0)
 117
 118
 119 if __name__ == u"__main__":
 120     main()