TCPDump Service#

TCPDump Service Class#

Classes:

TCPDump(hosts[, ifnames, networks, options, ...])

Monitor network traffic using tcpdump.

class enoslib.service.tcpdump.TCPDump(hosts: Iterable[Host], ifnames: List[str] | None = None, networks: Iterable[Network] | None = None, options: str = '', backup_dir: Path | str | None = None)#

Monitor network traffic using tcpdump.

This connects to every host to launch as many tcpdump processes as network interfaces to monitor. The pcap files can be retrieved and analysed by your favorite tool (wireshark, scappy …). Each tcpdump process is launched in the background using a dedicated tmux session.

Can be used as a Context Manager. In this case, the pcap files are retrieved automatically when exiting and all the remaining tcpdump processes are killed.

Note that if networks is used, sync_info() must have been called before.

Parameters:

hosts – list of hosts to consider
ifnames – explicit network card names to monitor. “any” is a possible keyword that will monitor all interfaces.
networks – monitor all interfaces that belong to one of those networks
options – extra options to pass to tcpdump command line.
backup_dir – path to a local directory where the pcap files will be saved

Examples

import logging
import tarfile
from pathlib import Path

from scapy.all import rdpcap

import enoslib as en

en.init_logging(level=logging.INFO)
en.check()


CLUSTER = "parasilo"
SITE = en.g5k_api_utils.get_cluster_site(CLUSTER)
job_name = Path(__file__).name

# claim the resources
conf = en.G5kConf.from_settings(job_name=job_name, walltime="0:20:00", job_type=[])
network = en.G5kNetworkConf(id="n1", type="prod", roles=["my_network"], site=SITE)
conf.add_network_conf(network).add_machine(
    roles=["control", "client"], cluster=CLUSTER, nodes=1, primary_network=network
).add_machine(
    roles=["control", "server"], cluster=CLUSTER, nodes=1, primary_network=network
).finalize()

provider = en.G5k(conf)
roles, networks = provider.init()

roles = en.sync_info(roles, networks)

# start a capture
# - on all the interface configured on the my_network network
# - we dump icmp traffic only
# - for the duration of the commands (here a client is pinging the server)
with en.TCPDump(
    hosts=roles["control"], networks=networks["my_network"], options="icmp"
) as t:
    backup_dir = t.backup_dir
    _ = en.run(f"ping -c10 {roles['server'][0].address}", roles["client"])

# pcap files are retrieved in the __enoslib__tcpdump__ directory
# - can be loaded in wireshark
# - manipulated with scappy ...


# Examples:
# create a dictionary of (alias, if) -> list of decoded packets by scapy
decoded_pcaps = {}
for host in roles["control"]:
    host_dir = backup_dir / host.alias
    t = tarfile.open(host_dir / "tcpdump.tar.gz")
    t.extractall(host_dir / "extracted")
    # get all extracted pcap for this host
    pcaps = (host_dir / "extracted").rglob("*.pcap")
    for pcap in pcaps:
        decoded_pcaps.setdefault(
            (host.alias, pcap.with_suffix("").name), rdpcap(str(pcap))
        )

# Displaying some packets
for (host, ifs), packets in decoded_pcaps.items():
    print(host, ifs)
    packets[0].show()
    packets[1].show()

backup(backup_dir: Path | None = None)#: (abstract) Backup the service.

deploy(force: bool = False)#: (abstract) Deploy the service.

destroy()#: (abstract) Destroy the service.