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:
8 # http://www.apache.org/licenses/LICENSE-2.0
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.
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.
22 from scapy.layers.inet import IP, UDP
23 from scapy.layers.l2 import Ether
24 from scapy.packet import Raw
26 from .PacketVerifier import RxQueue, TxQueue
27 from .TrafficScriptArg import TrafficScriptArg
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.
35 args = TrafficScriptArg(
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"
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")
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) /
63 pkt_raw = (Ether(src=tx_src_mac, dst=tx_dst_mac) /
64 IP(src=tx_src_ip, dst=tx_dst_ip) /
66 vxlan.VXLAN(vni=int(tx_vni)) /
70 # Send created packet on one interface and receive on the other
71 sent_packets.append(pkt_raw)
74 ether = rxq.recv(2, ignore=sent_packets)
76 # Check whether received packet contains layers Ether, IP and VXLAN
78 raise RuntimeError(u"Packet Rx timeout")
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:
87 f"VXLAN UDP port mismatch {ip.payload.dport} != 4789"
89 vxlan_pkt = ip.payload.payload
91 if int(vxlan_pkt.vni) != int(rx_vni):
92 raise RuntimeError(u"vxlan mismatch")
93 rx_pkt_p = vxlan_pkt.payload
95 if rx_pkt_p.src != tx_pkt_p.src:
97 f"RX encapsulated MAC src mismatch {rx_pkt_p.src} != {tx_pkt_p.src}"
99 if rx_pkt_p.dst != tx_pkt_p.dst:
101 f"RX encapsulated MAC dst mismatch {rx_pkt_p.dst} != {tx_pkt_p.dst}"
103 if rx_pkt_p[IP].src != tx_pkt_p[IP].src:
105 f"RX encapsulated IP src mismatch {rx_pkt_p[IP].src} != "
106 f"{tx_pkt_p[IP].src}"
108 if rx_pkt_p[IP].dst != tx_pkt_p[IP].dst:
110 f"RX encapsulated IP dst mismatch {rx_pkt_p[IP].dst} != "
111 f"{tx_pkt_p[IP].dst}"
114 # TODO: verify inner Ether()
118 if __name__ == u"__main__":