diff options
Diffstat (limited to 'resources/traffic_scripts/span_check.py')
| -rwxr-xr-x | resources/traffic_scripts/span_check.py | 128 |
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) |