1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
# SPDX-License-Identifier: BSD-3-Clause
# Copyright(c) 2010-2014 Intel Corporation
# Copyright(c) 2022 University of New Hampshire
# Copyright(c) 2023 PANTHEON.tech s.r.o.
"""Traffic generator node.
A traffic generator (TG) generates traffic that's sent towards the SUT node.
A TG node is where the TG runs.
"""
from scapy.packet import Packet # type: ignore[import-untyped]
from framework.config import TGNodeConfiguration
from framework.testbed_model.traffic_generator.capturing_traffic_generator import (
PacketFilteringConfig,
)
from .node import Node
from .port import Port
from .traffic_generator import CapturingTrafficGenerator, create_traffic_generator
class TGNode(Node):
"""The traffic generator node.
The TG node extends :class:`Node` with TG specific features:
* Traffic generator initialization,
* The sending of traffic and receiving packets,
* The sending of traffic without receiving packets.
Not all traffic generators are capable of capturing traffic, which is why there
must be a way to send traffic without that.
Attributes:
traffic_generator: The traffic generator running on the node.
"""
traffic_generator: CapturingTrafficGenerator
def __init__(self, node_config: TGNodeConfiguration):
"""Extend the constructor with TG node specifics.
Initialize the traffic generator on the TG node.
Args:
node_config: The TG node's test run configuration.
"""
super().__init__(node_config)
self.traffic_generator = create_traffic_generator(self, node_config.traffic_generator)
self._logger.info(f"Created node: {self.name}")
def send_packets_and_capture(
self,
packets: list[Packet],
send_port: Port,
receive_port: Port,
filter_config: PacketFilteringConfig = PacketFilteringConfig(),
duration: float = 1,
) -> list[Packet]:
"""Send `packets`, return received traffic.
Send `packets` on `send_port` and then return all traffic captured
on `receive_port` for the given duration. Also record the captured traffic
in a pcap file.
Args:
packets: The packets to send.
send_port: The egress port on the TG node.
receive_port: The ingress port in the TG node.
filter_config: The filter to use when capturing packets.
duration: Capture traffic for this amount of time after sending `packet`.
Returns:
A list of received packets. May be empty if no packets are captured.
"""
return self.traffic_generator.send_packets_and_capture(
packets,
send_port,
receive_port,
filter_config,
duration,
)
def send_packets(self, packets: list[Packet], port: Port):
"""Send packets without capturing resulting received packets.
Args:
packets: Packets to send.
port: Port to send the packets on.
"""
self.traffic_generator.send_packets(packets, port)
def close(self) -> None:
"""Free all resources used by the node.
This extends the superclass method with TG cleanup.
"""
self.traffic_generator.close()
super().close()
|
