import ipaddress
from scapy.layers.inet import IP, ICMP, ARP
-from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
+from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6ND_NS
from scapy.layers.l2 import Ether
from resources.libraries.python.PacketVerifier import RxQueue, TxQueue, auto_pad
"""Send a simple L2 or ICMP packet from one TG interface to DUT, then
receive a copy of the packet on the second TG interface, and a copy of
the ICMP reply."""
- args = TrafficScriptArg(
- ['tg_src_mac', 'src_ip', 'dst_ip', 'dut_if1_mac', 'ptype'])
+ args = TrafficScriptArg(['tg_src_mac', 'src_ip', 'dst_ip', 'dut_if1_mac',
+ 'ptype'])
src_mac = args.get_arg('tg_src_mac')
dst_mac = args.get_arg('dut_if1_mac')
rx_if = args.get_arg('rx_if')
ptype = args.get_arg('ptype')
- rxq = RxQueue(rx_if)
+ rxq_mirrored = RxQueue(rx_if)
+ rxq_tx = RxQueue(tx_if)
txq = TxQueue(tx_if)
+ sent = []
+
if ptype == "ARP":
pkt_raw = (Ether(src=src_mac, dst=dst_mac) /
ARP(hwsrc=src_mac, hwdst="00:00:00:00:00:00",
pkt_raw = (Ether(src=src_mac, dst=dst_mac) /
IP(src=src_ip, dst=dst_ip) /
ICMP(type="echo-request"))
- elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
+ else:
+ raise ValueError("IP addresses not in correct format")
+ elif ptype == "ICMPv6":
+ if valid_ipv6(src_ip) and valid_ipv6(dst_ip):
pkt_raw = (Ether(src=src_mac, dst=dst_mac) /
IPv6(src=src_ip, dst=dst_ip) /
ICMPv6EchoRequest())
else:
- raise ValueError("IP not in correct format")
+ raise ValueError("IPv6 addresses not in correct format")
else:
raise RuntimeError("Unexpected payload type.")
txq.send(pkt_raw)
- ether = rxq.recv(2)
+ sent.append(auto_pad(pkt_raw))
+
+ # Receive copy of Rx packet.
+ while True:
+ ether = rxq_mirrored.recv(2)
+ if ether is None:
+ raise RuntimeError("Rx timeout of mirrored Rx packet")
+
+ if ether.haslayer(ICMPv6ND_NS):
+ # read another packet in the queue if the current one is ICMPv6ND_NS
+ continue
+ else:
+ # otherwise process the current packet
+ break
- # Receive copy of sent packet.
- if ether is None:
- raise RuntimeError("Rx timeout")
pkt = auto_pad(pkt_raw)
if str(ether) != str(pkt):
- raise RuntimeError("Mirrored packet does not match packet sent.")
-
- # Receive copy of reply to sent packet.
- ether = rxq.recv(2)
+ print("Mirrored Rx packet doesn't match the original Rx packet.")
+ if ether.src != src_mac or ether.dst != dst_mac:
+ raise RuntimeError("MAC mismatch in mirrored Rx packet.")
+ if ptype == "ARP":
+ if not ether.haslayer(ARP):
+ raise RuntimeError("Mirrored Rx packet is not an ARP packet.")
+ if ether['ARP'].op != 1: # 1=who-has
+ raise RuntimeError("Mirrored Rx packet is not an ARP request.")
+ if ether['ARP'].hwsrc != src_mac or ether['ARP'].hwdst != dst_mac:
+ raise RuntimeError("MAC mismatch in mirrored Rx ARP packet.")
+ if ether['ARP'].psrc != src_ip or ether['ARP'].pdst != dst_ip:
+ raise RuntimeError("IP address mismatch in mirrored "
+ "Rx ARP packet.")
+ elif ptype == "ICMP":
+ if not ether.haslayer(IP):
+ raise RuntimeError("Mirrored Rx packet is not an IPv4 packet.")
+ if ether['IP'].src != src_ip or ether['IP'].dst != dst_ip:
+ raise RuntimeError("IP address mismatch in mirrored "
+ "Rx IPv4 packet.")
+ if not ether.haslayer(ICMP):
+ raise RuntimeError("Mirrored Rx packet is not an ICMP packet.")
+ if ether['ICMP'].type != 8: # 8=echo-request
+ raise RuntimeError("Mirrored Rx packet is not an ICMP "
+ "echo request.")
+ elif ptype == "ICMPv6":
+ if not ether.haslayer(IPv6):
+ raise RuntimeError("Mirrored Rx packet is not an IPv6 packet.")
+ if ether['IPv6'].src != src_ip or ether['IPv6'].dst != dst_ip:
+ raise RuntimeError("IP address mismatch in mirrored "
+ "Rx IPv6 packet.")
+ if not ether.haslayer(ICMPv6EchoRequest):
+ raise RuntimeError("Mirrored Rx packet is not an ICMPv6 "
+ "echo request.")
+ print("Mirrored Rx packet check OK.\n")
+
+ # Receive reply on TG Tx port.
+ ether_repl = rxq_tx.recv(2, sent)
+
+ if ether_repl is None:
+ raise RuntimeError("Reply not received on TG Tx port.")
+
+ print("Reply received on TG Tx port.\n")
+
+ # Receive copy of Tx packet.
+ ether = rxq_mirrored.recv(2)
if ether is None:
- raise RuntimeError("Rx timeout")
- if ether.src != dst_mac or ether.dst != src_mac:
- raise RuntimeError("MAC mismatch in mirrored response.")
- if ptype == "ARP":
- if ether['ARP'].op != 2:
- raise RuntimeError("Mirrored packet is not an ARP reply.")
- if ether['ARP'].hwsrc != dst_mac or ether['ARP'].hwdst != src_mac:
- raise RuntimeError("ARP MAC does not match l2 MAC "
- "in mirrored response.")
- if ether['ARP'].psrc != dst_ip or ether['ARP'].pdst != src_ip:
- raise RuntimeError("ARP IP address mismatch in mirrored response.")
- elif ptype == "ICMP" and ether.haslayer(IP):
- if ether['IP'].src != dst_ip or ether['IP'].dst != src_ip:
- raise RuntimeError("IP address mismatch in mirrored reply.")
- if ether['ICMP'].type != 0:
- raise RuntimeError("Mirrored packet is not an ICMP reply.")
- elif ptype == "ICMP" and ether.haslayer(IPv6):
- if ether['IPv6'].src != dst_ip or ether['IPv6'].dst != src_ip:
- raise RuntimeError("IP address mismatch in mirrored reply.")
- if not ether.haslayer(ICMPv6EchoReply):
- raise RuntimeError("Mirrored packet is not an ICMP reply.")
-
+ raise RuntimeError("Rx timeout of mirrored Tx packet")
+
+ if str(ether) != str(ether_repl):
+ print("Mirrored Tx packet doesn't match the received Tx packet.")
+ if ether.src != ether_repl.src or ether.dst != ether_repl.dst:
+ raise RuntimeError("MAC mismatch in mirrored Tx packet.")
+ if ptype == "ARP":
+ if not ether.haslayer(ARP):
+ raise RuntimeError("Mirrored Tx packet is not an ARP packet.")
+ if ether['ARP'].op != ether_repl['ARP'].op: # 2=is_at
+ raise RuntimeError("ARP operational code mismatch "
+ "in mirrored Tx packet.")
+ if ether['ARP'].hwsrc != ether_repl['ARP'].hwsrc\
+ or ether['ARP'].hwdst != ether_repl['ARP'].hwdst:
+ raise RuntimeError("MAC mismatch in mirrored Tx ARP packet.")
+ if ether['ARP'].psrc != ether_repl['ARP'].psrc\
+ or ether['ARP'].pdst != ether_repl['ARP'].pdst:
+ raise RuntimeError("IP address mismatch in mirrored "
+ "Tx ARP packet.")
+ elif ptype == "ICMP":
+ if not ether.haslayer(IP):
+ raise RuntimeError("Mirrored Tx packet is not an IPv4 packet.")
+ if ether['IP'].src != ether_repl['IP'].src\
+ or ether['IP'].dst != ether_repl['IP'].dst:
+ raise RuntimeError("IP address mismatch in mirrored "
+ "Tx IPv4 packet.")
+ if not ether.haslayer(ICMP):
+ raise RuntimeError("Mirrored Tx packet is not an ICMP packet.")
+ if ether['ICMP'].type != ether_repl['ICMP'].type: # 0=echo-reply
+ raise RuntimeError("ICMP packet type mismatch "
+ "in mirrored Tx packet.")
+ elif ptype == "ICMPv6":
+ if not ether.haslayer(IPv6):
+ raise RuntimeError("Mirrored Tx packet is not an IPv6 packet.")
+ if ether['IPv6'].src != ether_repl['IPv6'].src\
+ or ether['IPv6'].dst != ether_repl['IPv6'].dst:
+ raise RuntimeError("IP address mismatch in mirrored "
+ "Tx IPv6 packet.")
+ if ether[2].name != ether_repl[2].name:
+ raise RuntimeError("ICMPv6 message type mismatch "
+ "in mirrored Tx packet.")
+ print("Mirrored Tx packet check OK.\n")
sys.exit(0)
Code Review / csit.git / blobdiff