aboutsummaryrefslogtreecommitdiffstats
path: root/resources/traffic_scripts
diff options
context:
space:
mode:
Diffstat (limited to 'resources/traffic_scripts')
-rwxr-xr-xresources/traffic_scripts/ipfix_check.py16
-rwxr-xr-xresources/traffic_scripts/ipv4_sweep_ping.py12
-rwxr-xr-xresources/traffic_scripts/ipv6_sweep_ping.py24
3 files changed, 26 insertions, 26 deletions
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()