packet - Sending and capturing packets

Packet utilities for test suites.

The module provides helpers for:

  • Packet sending and verification,

  • Packet adjustments and modification.

Example

from scapy.layers.inet import IP
from scapy.layers.l2 import Ether
from api.packet import send_packet_and_capture, get_expected_packet, match_all_packets

pkt = Ether()/IP()/b"payload"
received = send_packet_and_capture(pkt)
expected = get_expected_packet(pkt)
match_all_packets([expected], received)
send_packet_and_capture(packet: scapy.packet.Packet, filter_config: PacketFilteringConfig = PacketFilteringConfig(no_lldp=True, no_arp=True, no_icmp=True), duration: float = 1) → list[scapy.packet.Packet]

Send and receive packet using the associated TG.

Send packet through the appropriate interface and receive on the appropriate interface. Modify the packet with l3/l2 addresses corresponding to the testbed and desired traffic.

Parameters:

  • packet (scapy.packet.Packet) – The packet to send.

  • filter_config (PacketFilteringConfig) – The filter to use when capturing packets.

  • duration (float) – Capture traffic for this amount of time after sending packet.

Returns:

A list of received packets.

Return type:

list[scapy.packet.Packet]

send_packets_and_capture(packets: list[scapy.packet.Packet], filter_config: PacketFilteringConfig = PacketFilteringConfig(no_lldp=True, no_arp=True, no_icmp=True), duration: float = 1) → list[scapy.packet.Packet]

Send and receive packets using the associated TG.

Send packets through the appropriate interface and receive on the appropriate interface. Modify the packets with l3/l2 addresses corresponding to the testbed and desired traffic.

Parameters:

  • packets (list[scapy.packet.Packet]) – The packets to send.

  • filter_config (PacketFilteringConfig) – The filter to use when capturing packets.

  • duration (float) – Capture traffic for this amount of time after sending packet.

Returns:

A list of received packets.

Return type:

list[scapy.packet.Packet]

send_packets(packets: list[scapy.packet.Packet]) → None

Send packets using the traffic generator and do not capture received traffic.

Parameters:

packets (list[scapy.packet.Packet]) – Packets to send.

get_expected_packets(packets: list[scapy.packet.Packet], sent_from_tg: bool = False) → list[scapy.packet.Packet]

Inject the proper L2/L3 addresses into packets.

Inject the L2/L3 addresses expected at the receiving end of the traffic generator.

Parameters:

  • packets (list[scapy.packet.Packet]) – The packets to modify.

  • sent_from_tg (bool) – If True packet was sent from the TG.

Returns:

packets with injected L2/L3 addresses.

Return type:

list[scapy.packet.Packet]

get_expected_packet(packet: scapy.packet.Packet, sent_from_tg: bool = False) → scapy.packet.Packet

Inject the proper L2/L3 addresses into packet.

Inject the L2/L3 addresses expected at the receiving end of the traffic generator.

Parameters:

  • packet (scapy.packet.Packet) – The packet to modify.

  • sent_from_tg (bool) – If True packet was sent from the TG.

Returns:

packet with injected L2/L3 addresses.

Return type:

scapy.packet.Packet

adjust_addresses(packets: list[scapy.packet.Packet], expected: bool = False) → list[scapy.packet.Packet]

L2 and L3 address additions in both directions.

Copies of packets will be made, modified and returned in this method.

Only missing addresses are added to packets, existing addresses will not be overridden. If any packet in packets has multiple IP layers (using GRE, for example) only the inner-most IP layer will have its addresses adjusted.

Assumptions:

Two links between SUT and TG, one link is TG -> SUT, the other SUT -> TG.

Parameters:

  • packets (list[scapy.packet.Packet]) – The packets to modify.

  • expected (bool) – If True, the direction is SUT -> TG, otherwise the direction is TG -> SUT.

Returns:

A list containing copies of all packets in packets after modification.

Raises:

InternalError – If no tests are running.

Return type:

list[scapy.packet.Packet]

match_all_packets(expected_packets: list[scapy.packet.Packet], received_packets: list[scapy.packet.Packet], verify: bool = True) → bool

Matches all the expected packets against the received ones.

Matching is performed by counting down the occurrences in a dictionary which keys are the raw packet bytes. No deep packet comparison is performed. All the unexpected packets (noise) are automatically ignored.

Parameters:

  • expected_packets (list[scapy.packet.Packet]) – The packets we are expecting to receive.

  • received_packets (list[scapy.packet.Packet]) – All the packets that were received.

  • verify (bool) – If True, and there are missing packets an exception will be raised.

Raises:

TestCaseVerifyError – if and not all the expected_packets were found in received_packets.

Returns:

True If there are no missing packets.

Return type:

bool

verify_packets(expected_packet: scapy.packet.Packet, received_packets: list[scapy.packet.Packet]) → None

Verify that expected_packet has been received.

Go through received_packets and check that expected_packet is among them. If not, raise an exception and log the last 10 commands executed on both the SUT and TG.

Parameters:

  • expected_packet (scapy.packet.Packet) – The packet we’re expecting to receive.

  • received_packets (list[scapy.packet.Packet]) – The packets where we’re looking for expected_packet.

Raises:

TestCaseVerifyError – expected_packet is not among received_packets.