3 # Copyright (c) 2020 Cisco and/or its affiliates.
5 # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
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:
12 # http://www.apache.org/licenses/LICENSE-2.0
13 # https://www.gnu.org/licenses/old-licenses/gpl-2.0-standalone.html
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 Apache 2.
19 # Unless required by applicable law or agreed to in writing, software
20 # distributed under the License is distributed on an "AS IS" BASIS,
21 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
22 # See the License for the specific language governing permissions and
23 # limitations under the License.
25 """Traffic script that sends an IP IPv4/IPv6 packet from one interface
26 to the other. Source and destination IP addresses and source and destination
27 MAC addresses are checked in received packet.
34 from robot.api import logger
35 from scapy.layers.inet import IP
36 from scapy.layers.inet6 import IPv6, ICMPv6ND_NS
37 from scapy.layers.l2 import Ether, Dot1Q
38 from scapy.packet import Raw
40 from .PacketVerifier import RxQueue, TxQueue
41 from .TrafficScriptArg import TrafficScriptArg
46 ipaddress.IPv4Address(ip)
48 except (AttributeError, ipaddress.AddressValueError):
54 ipaddress.IPv6Address(ip)
56 except (AttributeError, ipaddress.AddressValueError):
61 """Send IP/IPv6 packet from one traffic generator interface to the other."""
62 args = TrafficScriptArg(
64 u"tg_src_mac", u"tg_dst_mac", u"src_ip", u"dst_ip", u"dut_if1_mac",
68 u"encaps_tx", u"vlan_tx", u"vlan_outer_tx", u"encaps_rx",
69 u"vlan_rx", u"vlan_outer_rx"
73 tx_src_mac = args.get_arg(u"tg_src_mac")
74 tx_dst_mac = args.get_arg(u"dut_if1_mac")
75 rx_dst_mac = args.get_arg(u"tg_dst_mac")
76 rx_src_mac = args.get_arg(u"dut_if2_mac")
77 src_ip = args.get_arg(u"src_ip")
78 dst_ip = args.get_arg(u"dst_ip")
79 tx_if = args.get_arg(u"tx_if")
80 rx_if = args.get_arg(u"rx_if")
82 encaps_tx = args.get_arg(u"encaps_tx")
83 vlan_tx = args.get_arg(u"vlan_tx")
84 vlan_outer_tx = args.get_arg(u"vlan_outer_tx")
85 encaps_rx = args.get_arg(u"encaps_rx")
86 vlan_rx = args.get_arg(u"vlan_rx")
87 vlan_outer_rx = args.get_arg(u"vlan_outer_rx")
93 pkt_raw = Ether(src=tx_src_mac, dst=tx_dst_mac)
95 if encaps_tx == u"Dot1q":
96 pkt_raw /= Dot1Q(vlan=int(vlan_tx))
97 elif encaps_tx == u"Dot1ad":
99 pkt_raw /= Dot1Q(vlan=vlan_outer_tx)
100 pkt_raw /= Dot1Q(vlan=vlan_tx)
102 if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
103 pkt_raw /= IP(src=src_ip, dst=dst_ip, proto=61)
105 elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
106 pkt_raw /= IPv6(src=src_ip, dst=dst_ip)
109 raise ValueError(u"IP not in correct format")
112 sent_packets.append(pkt_raw)
117 ether = rxq.recv(2, ignore=sent_packets)
122 raise RuntimeError(u"IP packet Rx timeout")
124 if ether.haslayer(ICMPv6ND_NS):
125 # read another packet in the queue if the current one is ICMPv6ND_NS
128 # otherwise process the current packet
131 if rx_dst_mac == ether[Ether].dst and rx_src_mac == ether[Ether].src:
132 logger.trace(u"MAC matched")
134 raise RuntimeError(f"Matching packet unsuccessful: {ether!r}")
136 if encaps_rx == u"Dot1q":
137 if ether[Dot1Q].vlan == int(vlan_rx):
138 logger.trace(u"VLAN matched")
141 f"Ethernet frame with wrong VLAN tag "
142 f"({ether[Dot1Q].vlan}-received, "
143 f"{vlan_rx}-expected):\n{ether!r}"
145 ip = ether[Dot1Q].payload
146 elif encaps_rx == u"Dot1ad":
147 raise NotImplementedError()
151 if not isinstance(ip, ip_format):
152 raise RuntimeError(f"Not an IP packet received {ip!r}")
154 # Compare data from packets
156 logger.trace(u"Src IP matched")
159 f"Matching Src IP unsuccessful: {src_ip} != {ip.src}"
163 logger.trace(u"Dst IP matched")
166 f"Matching Dst IP unsuccessful: {dst_ip} != {ip.dst}"
172 if __name__ == u"__main__":