NICs like the Intel E810-C often produce ICMP packets. These packets
are stray and can interfere with testing. Therefore, add an ICMP
filtering option in the packet filter.
Moreover, use Scapy constants for the header values.
Signed-off-by: Luca Vizzarro <luca.vizza...@arm.com>
Reviewed-by: Paul Szczepanek <paul.szczepa...@arm.com>
---
Hi there,
sending in a new simple patch to introduce a new packet filter.
Best,
Luca
---
.../capturing_traffic_generator.py | 2 ++
.../testbed_model/traffic_generator/scapy.py | 20 +++++++++++++++++--
2 files changed, 20 insertions(+), 2 deletions(-)
diff --git
a/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
b/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
index e31ba2a9b7..61e5033f0b 100644
---
a/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
+++
b/dts/framework/testbed_model/traffic_generator/capturing_traffic_generator.py
@@ -34,10 +34,12 @@ class PacketFilteringConfig:
Attributes:
no_lldp: If :data:`True`, LLDP packets will be filtered out when
capturing.
no_arp: If :data:`True`, ARP packets will be filtered out when
capturing.
+ no_icmp: If :data:`True`, ICMP packets will be filtered out when
capturing.
"""
no_lldp: bool = True
no_arp: bool = True
+ no_icmp: bool = True
class CapturingTrafficGenerator(TrafficGenerator):
diff --git a/dts/framework/testbed_model/traffic_generator/scapy.py
b/dts/framework/testbed_model/traffic_generator/scapy.py
index 09adcafcd8..57d79aa3d4 100644
--- a/dts/framework/testbed_model/traffic_generator/scapy.py
+++ b/dts/framework/testbed_model/traffic_generator/scapy.py
@@ -19,6 +19,9 @@
from typing import ClassVar
from scapy.compat import base64_bytes
+from scapy.data import ETHER_TYPES, IP_PROTOS
+from scapy.layers.inet import IP
+from scapy.layers.inet6 import IPv6
from scapy.layers.l2 import Ether
from scapy.packet import Packet
@@ -234,10 +237,23 @@ def _set_packet_filter(self, filter_config:
PacketFilteringConfig):
def _filter(packet: Packet) -> bool:
if ether := packet.getlayer(Ether):
- if filter_config.no_arp and ether.type == 0x0806:
+ if filter_config.no_arp and ether.type == ETHER_TYPES.ARP:
return False
- if filter_config.no_lldp and ether.type == 0x88CC:
+ if filter_config.no_lldp and ether.type == ETHER_TYPES.LLDP:
+ return False
+
+ if ipv4 := packet.getlayer(IP):
+ if filter_config.no_icmp and ipv4.proto == IP_PROTOS.icmp:
+ return False
+
+ if ipv6 := packet.getlayer(IPv6):
+ next_header = ipv6.nh
+
+ if next_header == IP_PROTOS.hopopt:
+ next_header = ipv6.payload.nh
+
+ if filter_config.no_icmp and next_header ==
IP_PROTOS.ipv6_icmp:
return False
return True
--
2.43.0
