From 66919d66b8ab63369266eaab8db1628501e884aa Mon Sep 17 00:00:00 2001 From: Matej Klotton Date: Mon, 30 Jan 2017 12:36:34 +0100 Subject: Add pypcap python requirement Using pypcap's L2listen instead of standard AF_PACKET scapy can see received Dot1Q tag. Change-Id: Icb3dcb272a9611158a26a83fede7550bba3f367e Signed-off-by: Matej Klotton --- resources/traffic_scripts/ipfix_check.py | 16 ++++------------ resources/traffic_scripts/ipv4_sweep_ping.py | 12 ++++++++---- resources/traffic_scripts/ipv6_sweep_ping.py | 24 ++++++++++++++---------- 3 files changed, 26 insertions(+), 26 deletions(-) (limited to 'resources/traffic_scripts') diff --git a/resources/traffic_scripts/ipfix_check.py b/resources/traffic_scripts/ipfix_check.py index aa04b24038..2a08f0ce85 100755 --- a/resources/traffic_scripts/ipfix_check.py +++ b/resources/traffic_scripts/ipfix_check.py @@ -16,16 +16,16 @@ """Traffic script - IPFIX listener.""" import sys -from ipaddress import IPv4Address, IPv6Address, AddressValueError +from ipaddress import IPv4Address, IPv6Address, AddressValueError from scapy.layers.inet import IP, TCP, UDP from scapy.layers.inet6 import IPv6 from scapy.layers.l2 import Ether -from resources.libraries.python.telemetry.IPFIXUtil import IPFIXHandler, \ - IPFIXData from resources.libraries.python.PacketVerifier import RxQueue, TxQueue, auto_pad from resources.libraries.python.TrafficScriptArg import TrafficScriptArg +from resources.libraries.python.telemetry.IPFIXUtil import IPFIXHandler, \ + IPFIXData def valid_ipv4(ip): @@ -117,17 +117,10 @@ def main(): # allow scapy to recognize IPFIX headers and templates ipfix = IPFIXHandler() - - # clear receive buffer - while True: - pkt = rxq.recv(1, ignore=ignore, verbose=verbose) - if pkt is None: - break - data = None # get IPFIX template and data while True: - pkt = rxq.recv(5) + pkt = rxq.recv(10, ignore=ignore, verbose=verbose) if pkt is None: raise RuntimeError("RX timeout") if pkt.haslayer("IPFIXHeader"): @@ -194,5 +187,4 @@ def main(): if __name__ == "__main__": - main() diff --git a/resources/traffic_scripts/ipv4_sweep_ping.py b/resources/traffic_scripts/ipv4_sweep_ping.py index 7f6759724c..e258d45213 100755 --- a/resources/traffic_scripts/ipv4_sweep_ping.py +++ b/resources/traffic_scripts/ipv4_sweep_ping.py @@ -15,11 +15,12 @@ """Traffic script for IPv4 sweep ping.""" -import sys import logging import os +import sys + logging.getLogger("scapy.runtime").setLevel(logging.ERROR) -from resources.libraries.python.PacketVerifier import RxQueue, TxQueue,\ +from resources.libraries.python.PacketVerifier import RxQueue, TxQueue, \ create_gratuitous_arp_request, checksum_equal from resources.libraries.python.TrafficScriptArg import TrafficScriptArg from scapy.layers.inet import IP, ICMP @@ -54,7 +55,7 @@ def main(): # send ICMP echo request with incremented data length and receive ICMP # echo reply - for echo_seq in range(start_size, end_size+1, step): + for echo_seq in range(start_size, end_size + 1, step): pkt_send = (Ether(src=src_mac, dst=dst_mac) / IP(src=src_ip, dst=dst_ip) / ICMP(id=echo_id, seq=echo_seq) / @@ -83,7 +84,7 @@ def main(): if icmpv4.id != echo_id or icmpv4.seq != echo_seq: raise RuntimeError( - 'Invalid ICMP echo reply received ID {0} seq {1} should be ' \ + 'Invalid ICMP echo reply received ID {0} seq {1} should be ' 'ID {2} seq {3}, {0}'.format(icmpv4.id, icmpv4.seq, echo_id, echo_seq)) @@ -102,7 +103,10 @@ def main(): raise RuntimeError( 'Received ICMP payload does not match sent payload') + sent_packets.remove(pkt_send) + sys.exit(0) + if __name__ == "__main__": main() diff --git a/resources/traffic_scripts/ipv6_sweep_ping.py b/resources/traffic_scripts/ipv6_sweep_ping.py index da14c5d8ad..80fdda532a 100755 --- a/resources/traffic_scripts/ipv6_sweep_ping.py +++ b/resources/traffic_scripts/ipv6_sweep_ping.py @@ -15,11 +15,12 @@ """Traffic script for IPv6 sweep ping.""" -import sys import logging import os +import sys + logging.getLogger("scapy.runtime").setLevel(logging.ERROR) -from resources.libraries.python.PacketVerifier import RxQueue, TxQueue,\ +from resources.libraries.python.PacketVerifier import RxQueue, TxQueue, \ checksum_equal from resources.libraries.python.TrafficScriptArg import TrafficScriptArg from scapy.layers.inet6 import IPv6, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr @@ -51,19 +52,19 @@ def main(): # send ICMPv6 neighbor advertisement message sent_packets = [] pkt_send = (Ether(src=src_mac, dst=dst_mac) / - IPv6(src=src_ip, dst=dst_ip) / - ICMPv6ND_NA(tgt=src_ip, R=0) / - ICMPv6NDOptDstLLAddr(lladdr=src_mac)) + IPv6(src=src_ip, dst=dst_ip) / + ICMPv6ND_NA(tgt=src_ip, R=0) / + ICMPv6NDOptDstLLAddr(lladdr=src_mac)) sent_packets.append(pkt_send) txq.send(pkt_send) # send ICMPv6 echo request with incremented data length and receive ICMPv6 # echo reply - for echo_seq in range(start_size, end_size+1, step): + for echo_seq in range(start_size, end_size + 1, step): pkt_send = (Ether(src=src_mac, dst=dst_mac) / - IPv6(src=src_ip, dst=dst_ip) / - ICMPv6EchoRequest(id=echo_id, seq=echo_seq, - data=data[0:echo_seq])) + IPv6(src=src_ip, dst=dst_ip) / + ICMPv6EchoRequest(id=echo_id, seq=echo_seq, + data=data[0:echo_seq])) sent_packets.append(pkt_send) txq.send(pkt_send) @@ -88,7 +89,7 @@ def main(): if icmpv6.id != echo_id or icmpv6.seq != echo_seq: raise RuntimeError( - 'Invalid ICMPv6 echo reply received ID {0} seq {1} should be ' \ + 'Invalid ICMPv6 echo reply received ID {0} seq {1} should be ' 'ID {2} seq {3}, {0}'.format(icmpv6.id, icmpv6.seq, echo_id, echo_seq)) @@ -99,7 +100,10 @@ def main(): raise RuntimeError( 'Invalid checksum {0} should be {1}'.format(cksum, tmp.cksum)) + sent_packets.remove(pkt_send) + sys.exit(0) + if __name__ == "__main__": main() -- cgit 1.2.3-korg