aboutsummaryrefslogtreecommitdiffstats
path: root/resources/traffic_scripts
diff options
context:
space:
mode:
Diffstat (limited to 'resources/traffic_scripts')
-rwxr-xr-xresources/traffic_scripts/span_check.py128
1 files changed, 97 insertions, 31 deletions
diff --git a/resources/traffic_scripts/span_check.py b/resources/traffic_scripts/span_check.py
index 66400d9dfb..c7ca8a7374 100755
--- a/resources/traffic_scripts/span_check.py
+++ b/resources/traffic_scripts/span_check.py
@@ -75,9 +75,12 @@ def main():
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",
@@ -87,50 +90,113 @@ def main():
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))
+ ether = rxq_mirrored.recv(2)
- # Receive copy of sent packet.
+ # Receive copy of Rx packet.
if ether is None:
- raise RuntimeError("Rx timeout")
+ raise RuntimeError("Rx timeout of mirrored Rx packet")
pkt = auto_pad(pkt_raw)
if str(ether) != str(pkt):
- raise RuntimeError("Mirrored packet does not match packet sent.")
+ 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.")
+ else:
+ print("Reply received on TG Tx port.\n")
- # Receive copy of reply to sent packet.
- ether = rxq.recv(2)
+ # 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)