API Reference

Every public name, generated from the package's own docstrings.

The same surface in prose, with the bounds and the gotchas, is the API header that ships inside the installed package, pktcap/AGENTS.md, readable with importlib.resources.files("pktcap") and printed in the Guide of this site.

pktcap -- capture files, read, dissected, written and replayed.

Reads any pcap or pcapng capture, dissects each frame layer by layer with dissectors anyone can register (Ethernet, VLAN, IPv4, IPv6, UDP and TCP are built in), writes frames and datagrams back as pcap, pcapng or structured records, and replays a capture. What is not dissected is handed back as octets; what a protocol means stays with the library that registers its dissector.

Every public name is imported from :mod:pktcap; the modules beside this one are private.

CapturedDatagram

Bases: NamedTuple

One UDP datagram from a capture.

:ivar time: seconds since the epoch; for a reassembled datagram, the time of the fragment that completed it. :ivar source: (host, port) of the sender, the host as address text. :ivar destination: (host, port) it was sent to. :ivar payload: the octets after the UDP header. :ivar fragmented: payload is only what the first IP fragment carried. Set only by a decoder that was told not to reassemble. :ivar truncated: payload is shorter than the datagram's own length field says, because the capture's snap length cut the frame.

CapturedFrame

Bases: NamedTuple

One packet as a capture recorded it, before any dissection.

:ivar time: seconds since the epoch, as the capture states them. The file controls this value: it may be zero (a pcapng simple packet block has no timestamp) or far outside any calendar. :ivar linktype: the LINKTYPE_ number saying what data starts with. :ivar data: the captured octets, which a snap length may have cut short. :ivar interface: in a pcapng capture, the number of the interface the packet came from, counted from 0 within its section; None in a pcap capture, which has one interface and does not number it.

CopyResult

Bases: NamedTuple

What :func:copy_frames did.

:ivar read: items taken from the source. :ivar written: items the writer wrote. :ivar skipped: items select refused, or that had no datagram when datagrams was asked for. :ivar refused: items the writer turned away for its file budget during this call.

DissectedFrame

Bases: NamedTuple

A captured frame and the layers that were read from it.

:ivar frame: the frame as captured. :ivar layers: the layer records, outermost first. :ivar payloads: for each layer, the octets that follow it. The last one is what was not dissected. :ivar error: why dissection stopped early, naming the selector whose dissector could not read its layer; None when it simply ran out of dissectors or of octets. :ivar reassembled: the layers after an IP fragment were read from a datagram put together from several frames, of which this is the last.

time property

The frame's time, in seconds since the epoch.

payload property

The octets no dissector read: after the last layer, or the whole frame when there is none.

layer(kind)

The outermost layer that is a kind, or None.

Source code in src/pktcap/_dissect.py

def layer(self, kind: Type[_T]) -> Optional[_T]:
    """The outermost layer that is a ``kind``, or ``None``."""
    for layer in self.layers:
        if isinstance(layer, kind):
            return layer
    return None

payload_of(kind)

The octets that follow the outermost kind layer, or None.

Source code in src/pktcap/_dissect.py

def payload_of(self, kind: type) -> Optional[bytes]:
    """The octets that follow the outermost ``kind`` layer, or ``None``."""
    for layer, payload in zip(self.layers, self.payloads):
        if isinstance(layer, kind):
            return payload
    return None

datagram()

This frame as a UDP datagram: its time, both socket addresses and the octets after the UDP header, whatever dissected them further. None when the frame carries no UDP over IP.

Source code in src/pktcap/_dissect.py

def datagram(self) -> Optional[CapturedDatagram]:
    """This frame as a UDP datagram: its time, both socket addresses and
    the octets after the UDP header, whatever dissected them further.
    ``None`` when the frame carries no UDP over IP."""
    source = destination = None
    partial = False
    for layer, payload in zip(self.layers, self.payloads):
        if isinstance(layer, (IPv4Layer, IPv6Layer)):
            source, destination = layer.source, layer.destination
            partial = isinstance(layer, IPv4Layer) and layer.is_fragment
        elif isinstance(layer, IPv6FragmentLayer):
            partial = layer.is_fragment
        elif isinstance(layer, UDPLayer):
            if source is None or destination is None:
                return None
            fragmented = partial and not self.reassembled
            cut = layer.length > 8 + len(payload) and not fragmented
            return CapturedDatagram(
                self.frame.time,
                (source, layer.source_port),
                (destination, layer.destination_port),
                payload,
                fragmented,
                cut,
            )
    return None

DissectStats

Bases: NamedTuple

What a :class:FrameDissector has met in the frames it was given.

:ivar frames: frames given to dissect. :ivar malformed: frames in which a dissector could not read its layer (a ValueError), or that had more layers than the ceiling. :ivar failed: frames in which a dissector raised anything else, or returned something that is not a :class:Dissected: a defect in it. :ivar unsupported: frames whose link type has no dissector; :attr:FrameDissector.unsupported_linktypes says which. :ivar fragments: frames that were a piece of a fragmented IP datagram. :ivar dropped: reassemblies discarded: an overlap, a ceiling, old age. :ivar pending: reassemblies still waiting for a fragment.

FrameDissector(registry=None, *, reassemble=True, max_reassemblies=256, reassembly_timeout=30.0)

Dissects captured frames into :class:DissectedFrame objects.

Feed it every frame of a capture, in order: with reassemble on it keeps IP fragment state between frames. It raises nothing for a frame: every frame comes back, with the layers that were read and the rest as payload, and :attr:stats counts what went wrong. Not safe to share between threads.

:param registry: where dissectors are looked up; :func:default_registry when None. It is read at each frame, so a dissector registered later is used from then on. :param reassemble: put IP fragments back together. Off, no state is kept: the layers of a first fragment are read from the octets it carries, and a later fragment's octets are left as payload. :param max_reassemblies: the most datagrams being reassembled at once; one more discards the oldest. :param reassembly_timeout: seconds of capture time after which an unfinished datagram is discarded. :raises ValueError: a limit that is not positive.

Source code in src/pktcap/_dissect.py

def __init__(
    self,
    registry: Optional[DissectorRegistry] = None,
    *,
    reassemble: bool = True,
    max_reassemblies: int = 256,
    reassembly_timeout: float = 30.0,
) -> None:
    if registry is not None and not isinstance(registry, DissectorRegistry):
        raise TypeError("registry must be a DissectorRegistry")
    if isinstance(max_reassemblies, bool) or not isinstance(max_reassemblies, int):
        raise TypeError("max_reassemblies must be an int")
    if max_reassemblies < 1:
        raise ValueError("max_reassemblies must be at least 1")
    if not reassembly_timeout > 0:
        raise ValueError("reassembly_timeout must be positive")
    self._registry = registry if registry is not None else default_registry()
    self._reassemble = bool(reassemble)
    self._table = Reassembler(max_reassemblies, float(reassembly_timeout))
    self._frames = self._malformed = self._failed = 0
    self._unsupported = self._fragments = 0
    self._logged: Set[Selector] = set()
    self._by_linktype: Dict[int, int] = {}

registry property

The registry dissectors are looked up in.

stats property

The counters as they stand.

unsupported_linktypes property

How many frames each link type with no dissector had, by its LINKTYPE_ number: a read-only snapshot. The first 64 distinct link types are told apart; stats.unsupported counts every frame.

dissect(frame)

frame with every layer a registered dissector could read.

Source code in src/pktcap/_dissect.py

def dissect(self, frame: CapturedFrame) -> DissectedFrame:
    """``frame`` with every layer a registered dissector could read."""
    self._frames += 1
    lookup = self._registry.get
    selector: Selector = ("linktype", frame.linktype)
    dissector = lookup(*selector)
    if dissector is None:
        self._unsupported += 1
        counted = self._by_linktype
        if frame.linktype in counted or len(counted) < _MAX_COUNTED:
            counted[frame.linktype] = counted.get(frame.linktype, 0) + 1
        return DissectedFrame(frame, (), ())
    layers: List[object] = []
    payloads: List[bytes] = []
    data = frame.data
    error: Optional[str] = None
    reassembled = False
    steps = 0
    while dissector is not None:
        if steps == _MAX_LAYERS:
            self._malformed += 1
            error = "more than %d dissectors deep" % _MAX_LAYERS
            break
        steps += 1
        try:
            result = dissector(data)
            if not isinstance(result, Dissected) or not isinstance(
                result.payload, bytes
            ):
                raise TypeError("a dissector returns a Dissected with bytes")
        except ValueError as exc:
            self._malformed += 1
            error = "%s %d: %s" % (selector[0], selector[1], exc)
            break
        except Exception as exc:
            self._failed += 1
            error = "%s %d: the dissector raised %s" % (
                selector[0],
                selector[1],
                type(exc).__name__,
            )
            self._log_failure(selector, exc)
            break
        data = result.payload
        if result.layer is not None:
            layers.append(result.layer)
            payloads.append(data)
        if result.fragment is not None:
            self._fragments += 1
            if not self._reassemble:
                if result.fragment.offset:
                    break  # the middle of a datagram: no header to read
            else:
                whole = self._table.add(
                    self._unit(layers, result.fragment.key),
                    frame.time,
                    result.fragment.offset,
                    data,
                    result.fragment.last,
                )
                if whole is None:
                    break
                data, reassembled = whole, True
                if result.layer is not None:
                    payloads[-1] = whole
        dissector = None
        for selector in result.next:
            dissector = lookup(*selector)
            if dissector is not None:
                break
    return DissectedFrame(frame, tuple(layers), tuple(payloads), error, reassembled)

DissectorRegistry(*, builtins=True)

The dissectors one :class:FrameDissector chooses from.

A selector is a kind and a number. The built-in kinds are linktype (a capture's LINKTYPE_ number), ethertype, ip (an IP protocol number), udp and tcp (a port, the destination tried before the source). A dissector may name selectors of any other kind for what follows it, and whoever registers one of that kind is found.

Two registries share nothing, so two users in one process do not collide: build one, register into it, and pass it to :class:FrameDissector.

:param builtins: start with the built-in dissectors. False starts empty, for a caller that wants nothing decoded it did not choose.

Source code in src/pktcap/_dissectors/__init__.py

def __init__(self, *, builtins: bool = True) -> None:
    self._dissectors: Dict[Selector, Dissector] = {}
    self._layers: Dict[str, type] = {}
    self._layer_keys: Dict[str, Dict[str, KeyBuilder]] = {}
    if builtins:
        for table in (LINK_DISSECTORS, NETWORK_DISSECTORS, TRANSPORT_DISSECTORS):
            self._dissectors.update(table)

register(kind, value, dissector, *, replace=False)

Make dissector the one that reads what (kind, value) selects.

:param replace: take the place of the dissector already there. Without it a selector that is taken is an error: two libraries claiming one port should find out when they register, not when a capture decodes as the wrong protocol. :raises ValueError: the selector is taken and replace is false. :raises TypeError: dissector cannot be called.

Source code in src/pktcap/_dissectors/__init__.py

def register(
    self, kind: str, value: int, dissector: Dissector, *, replace: bool = False
) -> None:
    """Make ``dissector`` the one that reads what ``(kind, value)`` selects.

    :param replace: take the place of the dissector already there. Without
        it a selector that is taken is an error: two libraries claiming
        one port should find out when they register, not when a capture
        decodes as the wrong protocol.
    :raises ValueError: the selector is taken and ``replace`` is false.
    :raises TypeError: ``dissector`` cannot be called.
    """
    selector = _selector(kind, value)
    if not callable(dissector):
        raise TypeError("a dissector is a callable taking the layer's octets")
    if selector in self._dissectors and not replace:
        raise ValueError(
            "a dissector is already registered for %s %d; pass replace=True "
            "to take its place" % selector
        )
    self._dissectors[selector] = dissector

unregister(kind, value)

Forget the dissector for (kind, value); what it selected comes back undecoded. ValueError when there is none.

Source code in src/pktcap/_dissectors/__init__.py

def unregister(self, kind: str, value: int) -> None:
    """Forget the dissector for ``(kind, value)``; what it selected comes
    back undecoded. ``ValueError`` when there is none."""
    selector = _selector(kind, value)
    if selector not in self._dissectors:
        raise ValueError("no dissector is registered for %s %d" % selector)
    del self._dissectors[selector]

get(kind, value)

The dissector for (kind, value), or None: a selector with no dissector is the ordinary end of a dissection, not an error.

Source code in src/pktcap/_dissectors/__init__.py

def get(self, kind: str, value: int) -> Optional[Dissector]:
    """The dissector for ``(kind, value)``, or ``None``: a selector with
    no dissector is the ordinary end of a dissection, not an error."""
    return self._dissectors.get((kind, value))

selectors()

Every selector with a dissector, sorted.

Source code in src/pktcap/_dissectors/__init__.py

def selectors(self) -> Tuple[Selector, ...]:
    """Every selector with a dissector, sorted."""
    return tuple(sorted(self._dissectors))

register_layer(layer, *, name=None, keys=None, replace=False)

Declare the class of a layer a dissector returns, so a filter can name it, and optionally the filter keys that read it.

:param layer: the layer's class. Every field of a named tuple is a filter key NAME.FIELD with no further code. :param name: how a filter calls the layer: lower case, [a-z][a-z0-9_]*. Omitted: the class name in lower case without a trailing Layer, which is also how proto= and frame_record name it. :param keys: key name ([a-z][a-z0-9_-]*) to build(clause), called once per clause when a filter is compiled; it returns the test of one layer record and raises ValueError for a value that can never match. :param replace: take the place of a layer registered under that name. :raises ValueError: a name that is not valid, is a built-in layer's name, or is taken and replace is false; a key name that is not valid. :raises TypeError: layer is not a class, keys is not a mapping of callables.

Source code in src/pktcap/_dissectors/__init__.py

def register_layer(
    self,
    layer: type,
    *,
    name: Optional[str] = None,
    keys: Optional[Mapping[str, KeyBuilder]] = None,
    replace: bool = False,
) -> None:
    """Declare the class of a layer a dissector returns, so a filter can
    name it, and optionally the filter keys that read it.

    :param layer: the layer's class. Every field of a named tuple is a
        filter key ``NAME.FIELD`` with no further code.
    :param name: how a filter calls the layer: lower case, ``[a-z][a-z0-9_]*``.
        Omitted: the class name in lower case without a trailing
        ``Layer``, which is also how ``proto=`` and ``frame_record`` name it.
    :param keys: key name (``[a-z][a-z0-9_-]*``) to ``build(clause)``,
        called once per clause when a filter is compiled; it returns the
        test of one layer record and raises ``ValueError`` for a value
        that can never match.
    :param replace: take the place of a layer registered under that name.
    :raises ValueError: a name that is not valid, is a built-in layer's
        name, or is taken and ``replace`` is false; a key name that is
        not valid.
    :raises TypeError: ``layer`` is not a class, ``keys`` is not a
        mapping of callables.
    """
    if not isinstance(layer, type):
        raise TypeError("a layer is a class, such as a named tuple")
    if name is None:
        base = layer.__name__
        name = (
            base[:-5] if base.endswith("Layer") and len(base) > 5 else base
        ).lower()
    elif not isinstance(name, str):
        raise TypeError("a layer's name is text")
    if not _LAYER_NAME.match(name):
        raise ValueError(
            "%r is not a layer name (lower case letters, digits and _, "
            "starting with a letter): pass name=" % name
        )
    if name in BUILTIN_LAYERS:
        raise ValueError("%r is the name of a built-in layer" % name)
    if name in self._layers and not replace:
        raise ValueError(
            "a layer named %r is already registered; pass replace=True to "
            "take its place" % name
        )
    checked: Dict[str, KeyBuilder] = {}
    if keys is not None:
        if not isinstance(keys, Mapping):
            raise TypeError("keys maps a key name to a function building its test")
        for key, build in keys.items():
            if not isinstance(key, str) or not _KEY_NAME.match(key):
                raise ValueError(
                    "%r is not a key name (lower case letters, digits, _ and -, "
                    "starting with a letter, no dot)" % (key,)
                )
            if not callable(build):
                raise TypeError("the builder of key %r cannot be called" % key)
            checked[key] = build
    self._layers[name] = layer
    self._layer_keys[name] = checked

unregister_layer(name)

Forget a registered layer and its keys. ValueError when there is none, a built-in layer included.

Source code in src/pktcap/_dissectors/__init__.py

def unregister_layer(self, name: str) -> None:
    """Forget a registered layer and its keys. ``ValueError`` when there
    is none, a built-in layer included."""
    if not isinstance(name, str) or name not in self._layers:
        raise ValueError("no layer named %r is registered" % (name,))
    del self._layers[name]
    del self._layer_keys[name]

layers()

The registered layers, by name: a copy, built-in layers not in it.

Source code in src/pktcap/_dissectors/__init__.py

def layers(self) -> Dict[str, type]:
    """The registered layers, by name: a copy, built-in layers not in it."""
    return dict(self._layers)

copy()

A registry with the same dissectors, layers and keys that then changes on its own.

Source code in src/pktcap/_dissectors/__init__.py

def copy(self) -> "DissectorRegistry":
    """A registry with the same dissectors, layers and keys that then
    changes on its own."""
    clone = DissectorRegistry(builtins=False)
    clone._restore(self._snapshot())
    return clone

Dissected

Bases: NamedTuple

What a dissector returns for the octets it was given.

:ivar layer: the record of this layer: any immutable value, the built-in ones being named tuples. None for a step that only chooses what follows and has nothing to report. :ivar payload: the octets left for the next layer: a part of what the dissector was given, never more. :ivar next: the selectors to try for the next dissector, in order; the first one registered is used. Empty when nothing follows. :ivar fragment: set when payload is a piece of a larger unit.

Fragment

Bases: NamedTuple

Says that a dissector's payload is one piece of a larger unit.

The frame dissector puts the pieces together, within its bounds, and carries on with the whole once it is complete.

:ivar key: what identifies the unit: equal for every piece of it and for nothing else (addresses and an identification number, for IP). :ivar offset: where this piece's octets start in the unit. :ivar last: this piece ends the unit.

CaptureFilterError

Bases: PktcapError, ValueError

A capture-filter expression that cannot be compiled.

Raised when the filter is parsed or compiled, never while it is applied: a clause that is not key=value, an or, or a key or value that the protocol library's own builder refused (its ValueError is chained).

CaptureFormatError(message, *, offset=None)

Bases: PktcapError, ValueError

The input is not a pcap or pcapng capture, or it is a damaged one.

The one type every malformed container raises: a wrong magic number, a record or block cut short, a length over its ceiling, a block that contradicts itself. The message never quotes the file.

:ivar offset: how many octets of the input had been read when the problem was found, or None when that is not known.

Source code in src/pktcap/_exceptions.py

def __init__(self, message: str, *, offset: Optional[int] = None) -> None:
    if offset is not None:
        message = "%s (at octet %d)" % (message, offset)
    super().__init__(message)
    self.offset = offset

CaptureHookError(message, *, status=None, timed_out=False)

Bases: PktcapError, OSError

A hook command failed and was asked to end the capture (fail_fast).

An :class:OSError, as the failure of any other program or socket a capture relies on is. The message names the program and ends with a bounded, escaped tail of the program's error output.

:ivar status: the program's exit status, or None when it was killed for running past its time limit. :ivar timed_out: the program ran past its time limit and was killed with everything it had started.

Source code in src/pktcap/_exceptions.py

def __init__(
    self,
    message: str,
    *,
    status: Optional[int] = None,
    timed_out: bool = False,
) -> None:
    super().__init__(message)
    self.status = status
    self.timed_out = timed_out

CapturePluginError(message, *, plugin, source)

Bases: PktcapError, ValueError

A plugin the user named cannot be loaded.

Raised for a name that is no module, a module whose import failed, a module with no pktcap_plugin hook, a hook that takes no registry or raised, and an item of a list from a variable or a settings file that is no dotted Python name. The registry is left as it was before the call. A malformed item passed as an argument is a plain ValueError instead.

:ivar plugin: the item as written. :ivar source: where the list came from: "argument", or what the command line names ("PKTCAP_LOAD", a settings file's path).

Source code in src/pktcap/_exceptions.py

def __init__(self, message: str, *, plugin: str, source: str) -> None:
    super().__init__(message)
    self.plugin = plugin
    self.source = source

DissectError

Bases: PktcapError, ValueError

Octets that are not the layer a dissector was asked to read: a header cut short, or one whose length fields contradict what is there.

What the built-in dissectors raise. It never reaches a caller of :meth:pktcap.FrameDissector.dissect, which keeps the frame, leaves that layer undecoded and counts it; a caller that runs a dissector by hand catches it. A dissector of the caller's own may raise any ValueError.

LiveCaptureError

Bases: PktcapError, OSError

This platform cannot capture live: it has no AF_PACKET.

An :class:OSError, as the failure to open any other socket is. A process that lacks the capability gets the kernel's own PermissionError, which is not this class.

MissingExtraError(message, *, format, extra)

Bases: PktcapError, ImportError

An output format whose optional dependency is not installed.

An :class:ImportError, so except ImportError still catches it. The message names the extra to install; a caller that offers the format under its own extra words its own message from the attributes.

:ivar format: the format's name, as in OUTPUT_FORMATS ("toml"). :ivar extra: pktcap's extra that installs what the format needs.

Source code in src/pktcap/_exceptions.py

def __init__(self, message: str, *, format: str, extra: str) -> None:
    super().__init__(message)
    self.format = format
    self.extra = extra

PktcapError

Bases: Exception

The base of every exception pktcap raises on its own account.

RecordFormatError(message, *, format, lineno=None)

Bases: PktcapError, ValueError

Text that is not one record in the format it is read as.

Raised by :func:pktcap.loads_record for text that does not parse, that is not one document with a mapping at its top, that repeats a key, or that holds what a record cannot (a NaN in JSON, a tag that is not plain data, a nesting past the recursion limit). The message says what is wrong and where, and never quotes the text; the reader's own exception is not chained.

:ivar format: the format's name, as in RECORD_FORMATS ("yaml"). :ivar lineno: the 1-based line of the problem, or None when the reader keeps none.

Source code in src/pktcap/_exceptions.py

def __init__(
    self, message: str, *, format: str, lineno: Optional[int] = None
) -> None:
    super().__init__(
        message if lineno is None else "%s (at line %d)" % (message, lineno)
    )
    self._problem = message
    self.format = format
    self.lineno = lineno

UnsupportedFormatError

Bases: PktcapError, ValueError

No output format of that name, or none can be told from a file name.

The message lists the formats there are. A format that exists and whose extra is not installed is not this error: it raises :class:MissingExtraError.

FilterClause

Bases: NamedTuple

One key=value or key!=value clause of a capture filter.

:ivar key: the text before the operator, as written: case is kept. :ivar value: the text after it, surrounding space removed. :ivar negated: the operator was !=.

values property

value split on commas, each item stripped and empty ones left out, for a key that reads a comma as "any of".

EthernetLayer

Bases: NamedTuple

An Ethernet II header.

:ivar destination: the destination MAC, as aa:bb:cc:dd:ee:ff. :ivar source: the source MAC, in the same form. :ivar ethertype: what follows: 0x0800 IPv4, 0x86DD IPv6, 0x8100 a VLAN tag. Below 1,536 it is an 802.3 length, and nothing is selected to follow.

summary()

The addresses and what follows: aa:.. > bb:.. ethertype 0x0800.

Source code in src/pktcap/_layers.py

def summary(self) -> str:
    """The addresses and what follows: ``aa:.. > bb:.. ethertype 0x0800``."""
    return "%s > %s ethertype 0x%04x" % (
        self.source,
        self.destination,
        self.ethertype,
    )

IPv4Layer

Bases: NamedTuple

An IPv4 header. Options are skipped.

:ivar source: the source address, dotted. :ivar destination: the destination address. :ivar protocol: the protocol number of what follows (6 TCP, 17 UDP). :ivar ttl: the time to live. :ivar identification: the identification field, shared by the fragments of one datagram. :ivar dont_fragment: the DF flag. :ivar more_fragments: the MF flag: more fragments follow this one. :ivar fragment_offset: where this fragment's octets start in the datagram; 0 for a whole datagram and for a first fragment. :ivar length: the total length the header states, 0 when a network card that segments the send left it unset.

is_fragment property

Whether this packet is one piece of a fragmented datagram.

summary()

Addresses, protocol, time to live and identification, and where a fragment starts: 10.0.0.5 > 10.0.0.1 protocol 17 ttl 64 id 1.

Source code in src/pktcap/_layers.py

def summary(self) -> str:
    """Addresses, protocol, time to live and identification, and where a
    fragment starts: ``10.0.0.5 > 10.0.0.1 protocol 17 ttl 64 id 1``."""
    text = "%s > %s protocol %d ttl %d id %d" % (
        self.source,
        self.destination,
        self.protocol,
        self.ttl,
        self.identification,
    )
    if self.is_fragment:
        text += ", fragment at %d%s" % (
            self.fragment_offset,
            _more(self.more_fragments),
        )
    return text

IPv6ExtensionLayer

Bases: NamedTuple

A hop-by-hop (0), routing (43) or destination-options (60) header.

:ivar next_header: the protocol number of what follows. :ivar data: the header's octets after its first two.

summary()

What follows and how many octets the header holds past its first two.

Source code in src/pktcap/_layers.py

def summary(self) -> str:
    """What follows and how many octets the header holds past its first two."""
    return "next header %d, %d octets" % (self.next_header, len(self.data))

IPv6FragmentLayer

Bases: NamedTuple

An IPv6 fragment header.

:ivar next_header: the protocol number of the reassembled payload. :ivar fragment_offset: where this fragment's octets start, in octets. :ivar more_fragments: more fragments follow this one. :ivar identification: shared by the fragments of one datagram.

is_fragment property

False for an "atomic" fragment: a whole datagram in one piece.

summary()

What follows, the identification and where this piece starts.

Source code in src/pktcap/_layers.py

def summary(self) -> str:
    """What follows, the identification and where this piece starts."""
    return "next header %d, id %d, offset %d%s" % (
        self.next_header,
        self.identification,
        self.fragment_offset,
        _more(self.more_fragments),
    )

IPv6Layer

Bases: NamedTuple

An IPv6 fixed header. Extension headers are layers of their own.

:ivar source: the source address. A v4-mapped address is written ::ffff:10.0.0.5 on every Python. :ivar destination: the destination address. :ivar next_header: the protocol number of what follows. :ivar hop_limit: the hop limit. :ivar payload_length: the payload length the header states. :ivar traffic_class: the traffic class octet. :ivar flow_label: the 20-bit flow label.

summary()

Addresses, what follows and the hop limit.

Source code in src/pktcap/_layers.py

def summary(self) -> str:
    """Addresses, what follows and the hop limit."""
    return "%s > %s next header %d hop limit %d" % (
        self.source,
        self.destination,
        self.next_header,
        self.hop_limit,
    )

LinuxCookedLayer

Bases: NamedTuple

A Linux cooked-capture header, version 1 or 2: what the any device records in place of a link-layer header.

:ivar packet_type: 0 to this host, 1 broadcast, 2 multicast, 3 to another host, 4 sent by this host. :ivar hardware_type: the ARPHRD_ number of the device (1 Ethernet, 772 loopback). :ivar address: the sender's link-layer address as hex text, as many octets as the header says are valid (at most eight). :ivar ethertype: what follows. :ivar interface: the interface index; None in version 1, which has none.

summary()

Direction, device type, sender address when there is one, and what follows: host, hardware type 772, ethertype 0x0800.

Source code in src/pktcap/_layers.py

def summary(self) -> str:
    """Direction, device type, sender address when there is one, and what
    follows: ``host, hardware type 772, ethertype 0x0800``."""
    kind = _PACKET_TYPES.get(self.packet_type, "type %d" % self.packet_type)
    address = ", address %s" % self.address if self.address else ""
    return "%s, hardware type %d%s, ethertype 0x%04x" % (
        kind,
        self.hardware_type,
        address,
        self.ethertype,
    )

LoopbackLayer

Bases: NamedTuple

The four-octet header of a BSD loopback capture.

:ivar family: the address family as the capturing host numbered it: 2 is IPv4 everywhere; IPv6 is 10, 24, 28 or 30 by operating system.

summary()

The address family: family 2.

Source code in src/pktcap/_layers.py

def summary(self) -> str:
    """The address family: ``family 2``."""
    return "family %d" % self.family

TCPLayer

Bases: NamedTuple

A TCP header. The segment's payload follows it; :class:TCPReassembler puts the streams back together.

:ivar source_port: the source port. :ivar destination_port: the destination port. :ivar sequence: the sequence number, as on the wire. :ivar acknowledgment: the acknowledgment number, as on the wire. :ivar flags: the flag bits: 0x01 FIN, 0x02 SYN, 0x04 RST, 0x08 PSH, 0x10 ACK, 0x20 URG, 0x40 ECE, 0x80 CWR, 0x100 the accurate-ECN bit. :ivar window: the window field, unscaled. :ivar checksum: the checksum field, not verified. :ivar urgent: the urgent pointer. :ivar options: the option octets, undecoded.

syn property

The SYN flag: the segment opens a connection.

ack property

The ACK flag: the acknowledgment number means something.

fin property

The FIN flag: the sender has no more to send.

rst property

The RST flag: the connection is reset.

summary()

The ports, the flags that are set, the sequence and acknowledgment numbers and the window: 40000 > 80 [SYN,ACK] seq 100 ack 7 window 4096.

Source code in src/pktcap/_layers.py

def summary(self) -> str:
    """The ports, the flags that are set, the sequence and acknowledgment
    numbers and the window: ``40000 > 80 [SYN,ACK] seq 100 ack 7 window 4096``."""
    names = [name for bit, name in _TCP_FLAGS if self.flags & bit]
    return "%d > %d [%s] seq %d ack %d window %d" % (
        self.source_port,
        self.destination_port,
        ",".join(names) or "none",
        self.sequence,
        self.acknowledgment,
        self.window,
    )

UDPLayer

Bases: NamedTuple

A UDP header.

:ivar source_port: the source port. :ivar destination_port: the destination port. :ivar length: the length the header states, header included; 0 for a jumbogram or a send the network card segmented. More than the octets that follow means the capture cut the datagram short. :ivar checksum: the checksum field, not verified.

summary()

The ports and the stated length: 50000 > 69 length 13.

Source code in src/pktcap/_layers.py

def summary(self) -> str:
    """The ports and the stated length: ``50000 > 69 length 13``."""
    return "%d > %d length %d" % (
        self.source_port,
        self.destination_port,
        self.length,
    )

VLANLayer

Bases: NamedTuple

One 802.1Q or 802.1ad tag. A QinQ frame has two of these in a row.

:ivar id: the VLAN identifier, 0 to 4095. :ivar priority: the priority code point, 0 to 7. :ivar drop_eligible: the drop-eligible indicator. :ivar ethertype: what follows the tag.

summary()

The tag: vlan 5 priority 0 ethertype 0x0800.

Source code in src/pktcap/_layers.py

def summary(self) -> str:
    """The tag: ``vlan 5 priority 0 ethertype 0x0800``."""
    return "vlan %d priority %d%s ethertype 0x%04x" % (
        self.id,
        self.priority,
        " drop-eligible" if self.drop_eligible else "",
        self.ethertype,
    )

LiveCapture(interface=None, *, timeout=1.0)

A live capture of every packet on one interface, or on all of them.

Linux only. A context manager: constructing one opens nothing; open() or with opens the socket, which needs CAP_NET_RAW. Each packet comes back as a :class:CapturedFrame of link type 276 (Linux cooked capture v2, what tcpdump writes for its any device), stamped with the time it was read: ready for a :class:FrameDissector, and for a writer, whose output tcpdump and Wireshark read. On a loopback device, where every packet is seen leaving and arriving, only the arriving copy is returned.

:param interface: the interface to capture on: a name, a :class:netimps.Interface, or anything netimps.get_interface finds one by (an address, a MAC). None is every interface. :param timeout: how many seconds :meth:read waits for a packet. :raises ValueError: a timeout that is not positive.

Source code in src/pktcap/_live.py

def __init__(
    self, interface: InterfaceLike = None, *, timeout: float = 1.0
) -> None:
    if isinstance(timeout, bool) or not isinstance(timeout, (int, float)):
        raise TypeError("timeout must be a number of seconds")
    if not timeout > 0:
        raise ValueError("timeout must be positive")
    self._interface = interface
    self._timeout = float(timeout)
    self._socket: Optional[socket.socket] = None
    self._closed = False
    self._indexes: Dict[str, int] = {}

open()

Open the socket. Does nothing when it is already open.

:raises LiveCaptureError: this platform has no AF_PACKET. :raises PermissionError: the process lacks CAP_NET_RAW. :raises ValueError: no interface matches interface, or the capture was closed. :raises OSError: the kernel refused the interface.

Source code in src/pktcap/_live.py

def open(self) -> None:
    """Open the socket. Does nothing when it is already open.

    :raises LiveCaptureError: this platform has no ``AF_PACKET``.
    :raises PermissionError: the process lacks ``CAP_NET_RAW``.
    :raises ValueError: no interface matches ``interface``, or the capture
        was closed.
    :raises OSError: the kernel refused the interface.
    """
    if self._closed:
        raise ValueError("the capture is closed")
    if self._socket is not None:
        return
    # Before the interface is looked up: a platform with no capture says so
    # whatever name was given.
    _packet_family()
    name: Optional[str] = None
    if isinstance(self._interface, Interface):
        name = self._interface.name
    elif self._interface is not None:
        found = get_interface(self._interface)
        if found is None:
            raise ValueError("no interface matches %r" % (self._interface,))
        name = found.name
    self._socket = _open_socket(name)

read()

The next packet, or None when timeout passes without one.

:raises ValueError: the capture is not open.

Source code in src/pktcap/_live.py

def read(self) -> Optional[CapturedFrame]:
    """The next packet, or ``None`` when ``timeout`` passes without one.

    :raises ValueError: the capture is not open.
    """
    sock = self._socket
    if sock is None:
        raise ValueError("the capture is not open")
    deadline = time.monotonic() + self._timeout
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return None
        sock.settimeout(remaining)
        try:
            data, address = sock.recvfrom(_READ_SIZE)
        except socket.timeout:  # not the builtin TimeoutError before 3.10
            return None
        frame = self._frame(data, address, time.time())
        if frame is not None:
            return frame

fileno()

The socket's descriptor, for a caller's own event loop or selector.

:raises ValueError: the capture is not open.

Source code in src/pktcap/_live.py

def fileno(self) -> int:
    """The socket's descriptor, for a caller's own event loop or selector.

    :raises ValueError: the capture is not open.
    """
    if self._socket is None:
        raise ValueError("the capture is not open")
    return self._socket.fileno()

close()

Close the socket. Final, and harmless when repeated.

Source code in src/pktcap/_live.py

def close(self) -> None:
    """Close the socket. Final, and harmless when repeated."""
    self._closed = True
    sock, self._socket = self._socket, None
    if sock is not None:
        sock.close()

CaptureWriter(target, format=None, *, per_record=False, append=False, fields=(), max_files=1000)

Writes what was captured, or the records made of it, in one format.

What is written is a :class:CapturedDatagram (the UDP view) or a :class:DissectedFrame (anything). pcap and pcapng write the datagram under synthesised headers, or the frame as captured, and ignore the record. A record format writes the record given, or :func:datagram_record or :func:frame_record when none is given: json one line per record, yaml one document per record. toml and ini cannot hold two records in one file and need per_record=True. text writes :func:frame_summary of a frame, a line for a datagram, and ignores the record: many lines to a file, or one file a line.

Constructing a writer opens nothing: the file is opened by the first write. Not safe to share between threads.

:param target: a path or a binary stream; with per_record, a file-name pattern such as "cap/{timestamp}_{xid}.json". :param format: one of :data:OUTPUT_FORMATS. None takes it from the ending of target's name. :param per_record: write one file per record, named by the pattern. :param append: add to an existing file. Not for pcap or pcapng. :param fields: the pattern fields the caller will supply in names, on top of timestamp, index and format. :param max_files: the most files a per_record writer creates. A field value may come from the network, so a peer would otherwise choose how many files land on the disk. :raises UnsupportedFormatError: no such format, or none can be told. :raises MissingExtraError: the format's extra is not installed. :raises ValueError: a combination that cannot work, or a bad pattern.

Source code in src/pktcap/_output.py

def __init__(
    self,
    target: Union[str, "os.PathLike[str]", BinaryIO],
    format: Optional[str] = None,
    *,
    per_record: bool = False,
    append: bool = False,
    fields: Iterable[str] = (),
    max_files: int = 1000,
) -> None:
    if isinstance(target, io.TextIOBase):
        raise TypeError("target must be a binary stream, such as sys.stdout.buffer")
    if not isinstance(target, (str, os.PathLike)) and not hasattr(target, "write"):
        raise TypeError("target must be a path or a binary stream")
    if isinstance(max_files, bool) or not isinstance(max_files, int):
        raise TypeError("max_files must be an int")
    if max_files < 1:
        raise ValueError("max_files must be at least 1")
    self._format = infer_format(target, format)
    self._record_format: Optional[RecordFormat] = None
    self._line = self._format == TEXT_FORMAT
    if self._line:
        pass
    elif self._format not in CAPTURE_FORMATS:
        self._record_format = record_format(self._format)
        self._record_format.require()
        if not per_record and not self._record_format.streamable:
            raise ValueError(
                "%s cannot hold more than one record in a file: pass "
                "per_record=True and a file-name pattern, or use json or yaml"
                % self._format
            )
    elif append:
        raise ValueError("a %s capture cannot be appended to" % self._format)
    if append and per_record:
        raise ValueError(
            "append adds to one growing file: it cannot be used with per_record"
        )
    self._fields = tuple(fields)
    self._pattern: Optional[str] = None
    if per_record:
        if not isinstance(target, (str, os.PathLike)):
            raise ValueError("per_record needs a file-name pattern, not a stream")
        self._pattern = self._checked_pattern(os.fspath(target))
    self._target = target
    self._append = bool(append)
    self._max_files = max_files
    self._file: Optional[BinaryIO] = None
    self._capture: Optional[Union[PcapWriter, PcapngWriter]] = None
    self._paths: Set[str] = set()
    self._closed = False
    self._warned = False
    self._count = 0
    #: Frames, datagrams or records written.
    self.written = 0
    #: Records not written because ``max_files`` had been reached.
    self.refused = 0

format property

The name of the format being written.

write(item, record=None, *, names=None)

Write one datagram or frame, or the record made of it.

:param item: what was captured: a :class:CapturedDatagram or a :class:DissectedFrame. A capture format writes it; a record format reads its time for {timestamp}. :param record: plain data for a record format. None writes :func:datagram_record or :func:frame_record of item. :param names: with per_record, a value for each of fields. :raises ValueError: a closed writer, a field with no value, or an item or record the format must refuse. :raises TypeError: an item of another type, or a record the format cannot represent. :raises OSError: the file cannot be opened or written.

Source code in src/pktcap/_output.py

def write(
    self,
    item: Union[CapturedDatagram, DissectedFrame],
    record: Optional[Mapping[str, Any]] = None,
    *,
    names: Optional[Mapping[str, object]] = None,
) -> None:
    """Write one datagram or frame, or the record made of it.

    :param item: what was captured: a :class:`CapturedDatagram` or a
        :class:`DissectedFrame`. A capture format writes it; a record
        format reads its time for ``{timestamp}``.
    :param record: plain data for a record format. ``None`` writes
        :func:`datagram_record` or :func:`frame_record` of ``item``.
    :param names: with ``per_record``, a value for each of ``fields``.
    :raises ValueError: a closed writer, a field with no value, or an item
        or record the format must refuse.
    :raises TypeError: an item of another type, or a record the format
        cannot represent.
    :raises OSError: the file cannot be opened or written.
    """
    if self._closed:
        raise ValueError("the writer is closed")
    if not isinstance(item, (CapturedDatagram, DissectedFrame)):
        raise TypeError("item must be a CapturedDatagram or a DissectedFrame")
    data = b""
    if self._line:
        line = frame_summary(item) if isinstance(item, DissectedFrame) else None
        if line is None:
            line = datagram_summary(item)  # type: ignore[arg-type]
        data = (line + "\n").encode("ascii")
    elif self._record_format is not None:
        if record is None:
            if isinstance(item, DissectedFrame):
                record = frame_record(item)
            else:
                record = datagram_record(item)
        elif not isinstance(record, Mapping):
            raise TypeError("a record is a mapping")
        data = self._record_format.dumps(record).encode("utf-8")
    elif self._pattern is not None:
        buffer = io.BytesIO()
        self._capture_write(self._capture_writer(buffer), item)
        data = buffer.getvalue()
    if self._pattern is not None:
        index, self._count = self._count, self._count + 1
        self._write_file(self._pattern, item.time, data, index, names or {})
    elif self._record_format is None and not self._line:
        if self._capture is None:
            self._make_parents(self._target)
            self._capture = self._capture_writer(self._target)
        self._capture_write(self._capture, item)
        self.written += 1
    else:
        separator = b""
        if self._record_format is not None:
            separator = self._record_format.separator.encode("ascii")
        self._write_stream(separator + data)

close()

Close the file if this writer opened it. Harmless when repeated.

Source code in src/pktcap/_output.py

def close(self) -> None:
    """Close the file if this writer opened it. Harmless when repeated."""
    self._closed = True
    opened, self._file = self._file, None
    if opened is not None and isinstance(self._target, (str, os.PathLike)):
        opened.close()
    capture, self._capture = self._capture, None
    if capture is not None:
        capture.close()

LoadedPlugin

Bases: NamedTuple

One plugin :func:load_plugins loaded.

:ivar name: the item as written. :ivar source: where the list came from: "argument" for the one :func:load_plugins was given, or whatever the command line names (a variable, a settings file's path); "always" for a plugin the caller's code loads whatever the user listed. :ivar selectors: the selectors its hook registered a dissector under. :ivar layers: the names of the layers its hook declared.

ReplayResult

Bases: NamedTuple

What :func:replay_to did.

:ivar sent: datagrams sent. :ivar partial: datagrams not sent because they were marked fragmented or truncated: their payload is not the whole message.

TCPReassembler(*, max_streams=1024, max_buffered=16777216, idle_timeout=300.0)

Bases: Engine

Puts TCP streams back together from dissected frames.

Feed it every frame of a capture, in capture order. It raises nothing for a frame and keeps its state within the bounds below. Not safe to share between threads.

:param max_streams: the most connections tracked; one more forgets the least recently active, whose held octets are dropped. :param max_buffered: the most octets held out of order over all connections, each held piece charged its length plus 64. Past it, the direction that has waited longest gives up its holes and delivers. :param idle_timeout: seconds of capture time after which a silent connection is forgotten when its addresses are next seen; they then start a new stream. :raises TypeError: an option of the wrong type. :raises ValueError: a limit that is not positive.

Source code in src/pktcap/_streams/_tcp.py

def __init__(
    self,
    *,
    max_streams: int = 1024,
    max_buffered: int = 16777216,
    idle_timeout: float = 300.0,
) -> None:
    for name, value in (
        ("max_streams", max_streams),
        ("max_buffered", max_buffered),
    ):
        if isinstance(value, bool) or not isinstance(value, int):
            raise TypeError("%s must be an int" % name)
        if value < 1:
            raise ValueError("%s must be at least 1" % name)
    if isinstance(idle_timeout, bool) or not isinstance(idle_timeout, (int, float)):
        raise TypeError("idle_timeout must be a number")
    if not idle_timeout > 0:
        raise ValueError("idle_timeout must be positive")
    try:
        idle = float(idle_timeout)
    except OverflowError:
        raise ValueError("idle_timeout is too large for a float") from None
    super().__init__(max_buffered)
    self._max_streams = max_streams
    self._idle = idle
    self._table: "OrderedDict[_Key, _Connection]" = OrderedDict()
    # Connections every direction of which has ended, least recently
    # active first: they give up their place before any other.
    self._done: "OrderedDict[_Key, None]" = OrderedDict()

add(frame)

One frame; what it made deliverable, in the order it became so.

A frame can release the other direction's held octets, on an acknowledgment, before it delivers its own. A frame with no TCP over IP gives ().

Source code in src/pktcap/_streams/_tcp.py

def add(self, frame: DissectedFrame) -> Tuple[TCPStreamData, ...]:
    """One frame; what it made deliverable, in the order it became so.

    A frame can release the other direction's held octets, on an
    acknowledgment, before it delivers its own. A frame with no TCP over
    IP gives ``()``.
    """
    if not isinstance(frame, DissectedFrame):
        raise TypeError("frame must be a DissectedFrame")
    found = _segment_of(frame)
    if found is None:
        return ()
    tcp, source, destination, data, fragment, snapped = found
    self._segments += 1
    if fragment:
        self._ignored += 1
        return ()
    time = frame.time
    if time != time:  # not a number: says nothing, as a time of 0.0 does
        time = 0.0
    if time != 0.0:  # a pcapng simple packet block has no time
        self._now = time
    key: _Key = (
        (source, destination) if source < destination else (destination, source)
    )
    out: List[TCPStreamData] = []
    conn = self._lookup(key, time)
    if conn is None and (tcp.rst or not (data or tcp.syn or tcp.fin)):
        return ()
    if conn is not None and tcp.rst:
        # A reset from a side that has sent nothing is believed whatever
        # its number.
        sender = conn.directions.get(source)
        if sender is not None:
            start = sender.offset_of(tcp.sequence)
            behind = sender.behind(tcp.sequence)
            if behind:
                self._ignored += 1
            if behind or (
                start > sender.top + FAR
                and self._set_aside(sender, start, start, time, out, False)
            ):
                self._touch(key, conn, time)
                return ()
        self._replace(key, conn, time, out)
        return tuple(out)
    aside = conn.aside if conn is not None else None
    if conn is not None and aside is not None:
        if self._confirms(aside, source, tcp):
            origin, target, number = aside
            self._ignored -= 1
            self._replace(key, conn, time, out)
            conn = self._create(key)
            conn.directions[origin] = Direction(
                origin, target, conn.stream, self._shared, number, True
            )
        elif source == aside[0] and not tcp.syn and (data or tcp.fin):
            # Its sender went on from somewhere else: the SYN was not its
            # own. Kept, a stream that reached that number much later
            # would confirm it.
            conn.aside = None
    if conn is not None and tcp.syn:
        verdict = self._syn(conn, source, destination, tcp)
        if verdict == _IGNORE:
            self._ignored += 1
            return tuple(out)
        if verdict == _NEW:
            if any(not d.ended for d in conn.directions.values()):
                conn.aside = (source, destination, tcp.sequence)
                self._ignored += 1
                return tuple(out)
            self._replace(key, conn, time, out)
            conn = None
    if conn is None:
        conn = self._create(key)
    self._touch(key, conn, time)
    if tcp.ack:
        self._acknowledge(
            conn.directions.get(destination), tcp.acknowledgment, time, out
        )
    if data or tcp.syn or tcp.fin:
        direction = conn.directions.get(source)
        if direction is None:
            direction = Direction(
                source,
                destination,
                conn.stream,
                self._shared,
                tcp.sequence,
                tcp.syn,
            )
            conn.directions[source] = direction
        self._segment(direction, tcp, data, snapped, time, out)
    self._enforce(time, out)
    if key in self._table:
        self._note(key, conn)
    return tuple(out)

flush()

The end of the capture: everything still held, each run after a hole with its missing, connections in order of stream. The table is empty afterwards. A FIN that was captured still ends its direction; flushing marks no other end.

Source code in src/pktcap/_streams/_tcp.py

def flush(self) -> Tuple[TCPStreamData, ...]:
    """The end of the capture: everything still held, each run after a
    hole with its ``missing``, connections in order of ``stream``. The
    table is empty afterwards. A FIN that was captured still ends its
    direction; flushing marks no other end."""
    out: List[TCPStreamData] = []
    for conn in sorted(self._table.values(), key=lambda item: item.stream):
        for direction in conn.directions.values():
            if not direction.ended:
                self._give_up(direction, self._now, out)
    self._table.clear()
    self._done.clear()
    self._waiting.clear()
    return tuple(out)

TCPStreamData

Bases: NamedTuple

Octets of one direction of a TCP connection, in order.

:ivar time: the capture time of the frame that made the octets deliverable. :ivar source: the sender's (host, port). :ivar destination: the receiver's (host, port). :ivar data: the octets; empty only when end is true. :ivar offset: where data[0] lies in this direction, counted from 0 and never wrapping; octets given up on count. :ivar missing: the octets given up on immediately before data; they are already counted in offset. :ivar stream: the connection's number, shared by both directions, from 0 in the order connections first appeared. :ivar end: this direction is over after data. When octets were given up between the last ones handed out and the end, the end is an item of its own, with no data and those octets in missing.

TCPStreamStats

Bases: NamedTuple

What a :class:TCPReassembler has met in the segments it was given.

:ivar segments: TCP segments taken, whatever became of them. :ivar streams: connections numbered so far. :ivar delivered: octets handed out. :ivar retransmitted: octets dropped as copies of delivered or held ones, and octets captured once after their place was given up. :ivar out_of_order: segments that arrived ahead of a hole. :ivar missing: octets given up on. :ivar conflicts: held octets that a later copy disagreed with. :ivar ignored: segments and acknowledgments set aside: octets before the start or beyond a FIN, an out-of-sequence RST, a segment, FIN or RST too far ahead, a SYN not yet confirmed, a SYN-ACK that contradicts the connection, an acknowledgment beyond everything believed, a segment in an IP fragment that was not reassembled. :ivar evicted: connections forgotten at max_streams or for age; their held octets are dropped. A connection every direction of which had ended gives up its place first and is never counted. :ivar pending: connections in the table. :ivar held: octets held out of order at the moment. :ivar dropped: octets held when their connection was forgotten.

UDPCapture(endpoints, *, timeout=1.0, max_size=_MAX_SIZE)

Datagrams arriving at sockets the caller bound, read as raw-IP frames.

The twin of :class:LiveCapture on every platform and with no privilege. The frames have link type 101 and a made-up IP header: the addresses are the sender's and the receiver's as the socket reports them (the destination is the address the datagram was sent to when the host says so, else the address the socket is bound to, which may be a wildcard), the time is when it was read and interface the arrival interface when the host reports one. The capture owns the endpoints from construction: it closes them, once, in :meth:close. Not safe to share between threads.

:param endpoints: bound :class:netimps.UDPEndpoint objects, at most 256. An endpoint limited to interfaces (UDPEndpoint(interfaces=)) is asked about every datagram: one it does not admit is counted in :attr:not_admitted and dropped before a frame is built. :param timeout: how many seconds :meth:read waits for a datagram. :param max_size: the largest payload returned; a bigger datagram is counted in :attr:truncated and dropped, since what a socket returns of it is only its start. :raises TypeError: an endpoint that is not a UDPEndpoint, or an option of the wrong type. The endpoints are then still the caller's. :raises ValueError: no endpoint, more than 256, an endpoint given twice, a timeout that is not positive or a max_size outside 1 to 65535.

Source code in src/pktcap/_sources/_udp.py

def __init__(
    self,
    endpoints: Iterable[UDPEndpoint],
    *,
    timeout: float = 1.0,
    max_size: int = _MAX_SIZE,
) -> None:
    held = tuple(itertools.islice(endpoints, MAX_ENDPOINTS + 1))
    if not held:
        raise ValueError("a UDP capture needs at least one endpoint")
    if len(held) > MAX_ENDPOINTS:
        raise ValueError("a UDP capture reads at most %d endpoints" % MAX_ENDPOINTS)
    for endpoint in held:
        if not isinstance(endpoint, UDPEndpoint):
            raise TypeError("endpoints must be netimps.UDPEndpoint objects")
    if len({id(endpoint.socket) for endpoint in held}) != len(held):
        raise ValueError("an endpoint is given twice")
    if not _number(timeout, "timeout") > 0:
        raise ValueError("timeout must be positive")
    if isinstance(max_size, bool) or not isinstance(max_size, int):
        raise TypeError("max_size must be an int")
    if not 1 <= max_size <= _MAX_SIZE:
        raise ValueError("max_size must be from 1 to %d" % _MAX_SIZE)
    self._endpoints = held
    self._timeout = float(timeout)
    self._max_size = max_size
    self._closed = False
    self._cursor = 0
    self._selector: Optional[selectors.BaseSelector] = None
    self._queue: Optional["asyncio.Queue[object]"] = None
    self._readers: List["asyncio.Future[None]"] = []
    #: Datagrams over ``max_size`` (or that no frame can hold), read and
    #: dropped.
    self.truncated = 0
    #: Datagrams an endpoint does not admit (``UDPEndpoint.admits``: they
    #: arrived on an interface it does not serve), read and dropped.
    self.not_admitted = 0

read()

The next frame, or None when timeout seconds pass without one. A datagram over max_size is counted and skipped.

:raises ValueError: the capture is closed. :raises OSError: a socket failed.

Source code in src/pktcap/_sources/_udp.py

def read(self) -> Optional[CapturedFrame]:
    """The next frame, or ``None`` when ``timeout`` seconds pass without
    one. A datagram over ``max_size`` is counted and skipped.

    :raises ValueError: the capture is closed.
    :raises OSError: a socket failed.
    """
    if self._closed:
        raise ValueError("the capture is closed")
    deadline = time.monotonic() + self._timeout
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return None
        for endpoint in self._select(remaining):
            datagram = self._receive(endpoint)
            if datagram is None:
                continue
            self._cursor = (self._endpoints.index(endpoint) + 1) % len(
                self._endpoints
            )
            frame = self._frame(endpoint, datagram)
            if frame is not None:
                return frame
            break

aread() async

:meth:read, awaited: one task for each endpoint feeds a bounded queue, started by the first call, and the next frame is taken from it.

Use one event loop for a capture's life, and :meth:aclose from a coroutine. Do not mix it with :meth:read.

:raises ValueError: the capture is closed. :raises OSError: a socket failed.

Source code in src/pktcap/_sources/_udp.py

async def aread(self) -> Optional[CapturedFrame]:
    """:meth:`read`, awaited: one task for each endpoint feeds a bounded
    queue, started by the first call, and the next frame is taken from it.

    Use one event loop for a capture's life, and :meth:`aclose` from a
    coroutine. Do not mix it with :meth:`read`.

    :raises ValueError: the capture is closed.
    :raises OSError: a socket failed.
    """
    import asyncio

    if self._closed:
        raise ValueError("the capture is closed")
    if self._queue is None:
        queue: "asyncio.Queue[object]" = asyncio.Queue(_QUEUE)
        self._queue = queue
        self._readers = [
            asyncio.ensure_future(self._pump(endpoint, queue))
            for endpoint in self._endpoints
        ]
    item = await self._next(self._queue)
    if isinstance(item, BaseException):
        raise item
    return item if isinstance(item, CapturedFrame) else None

close()

Close every endpoint. Final, and harmless when repeated.

From a coroutine use :meth:aclose: this waits for netimps' reader thread, if the asynchronous read started one.

Source code in src/pktcap/_sources/_udp.py

def close(self) -> None:
    """Close every endpoint. Final, and harmless when repeated.

    From a coroutine use :meth:`aclose`: this waits for netimps' reader
    thread, if the asynchronous read started one.
    """
    self._closed = True
    self._stop_readers()
    selector, self._selector = self._selector, None
    if selector is not None:
        selector.close()
    for endpoint in self._endpoints:
        endpoint.close()

aclose() async

:meth:close, awaited: the readers are cancelled and have left, and the endpoints are closed, when it returns. Harmless when repeated.

Source code in src/pktcap/_sources/_udp.py

async def aclose(self) -> None:
    """:meth:`close`, awaited: the readers are cancelled and have left, and
    the endpoints are closed, when it returns. Harmless when repeated."""
    import asyncio

    self._closed = True
    try:
        readers = self._stop_readers()
        if readers:
            await asyncio.gather(*readers, return_exceptions=True)
        for endpoint in self._endpoints:
            await endpoint.aclose()
    finally:
        self.close()

PcapngWriter(target)

Bases: _CaptureFileWriter

Writes a pcapng file that tcpdump and Wireshark read.

One section, little-endian, microsecond timestamps, one interface description for each link type in the order they are first written, and an enhanced packet block for each frame or datagram. Unlike pcap, one file holds frames of any mix of link types, so any capture can be written back. No option is written: nothing about the host, the tool or the interfaces beyond their link types.

The lifecycle is :class:PcapWriter's: nothing is opened until the first write, each block is flushed, a stream stays the caller's.

:param target: a path, or a binary stream.

Source code in src/pktcap/_writer.py

def __init__(self, target: Union[str, "os.PathLike[str]", BinaryIO]) -> None:
    super().__init__(target)
    self._interfaces: Dict[int, int] = {}

PcapWriter(target)

Bases: _CaptureFileWriter

Writes a pcap file that tcpdump and Wireshark read.

Little-endian, microsecond pcap. A pcap file has one link type, fixed by the first thing written: raw IP (101) for a datagram, the frame's own for a frame. A later write of another link type is refused; a capture of several link types is written with :class:PcapngWriter.

For a datagram, a v4-mapped IPv6 address is written as IPv4, which is what was on the wire when a dual-stack socket reported it; a datagram with one IPv4 and one IPv6 end is written as IPv6.

Constructing a writer touches nothing: a path is opened (and an existing file replaced), and the file header written, by the first write. A writer that never writes creates no file. Each record is flushed, so a capture can be followed while it grows. Not safe to share between threads.

:param target: a path, or a binary stream. A stream stays the caller's to close; a path this writer opened is closed by :meth:close.

Source code in src/pktcap/_writer.py

def __init__(self, target: Union[str, "os.PathLike[str]", BinaryIO]) -> None:
    super().__init__(target)
    self._linktype: Optional[int] = None

read_frames(source, *, max_frame_size=_DEFAULT_MAX_FRAME_SIZE)

Every packet of a pcap or pcapng capture, in file order.

Nothing is read until the first frame is asked for, and frames are yielded as they are read, so a live pipe (sys.stdin.buffer) works and a damaged capture gives every frame before the damage and then raises. An empty input is an empty capture.

:param source: a path, or a binary stream, which need not be seekable. A stream stays the caller's to close. :param max_frame_size: the most octets one captured frame may have. A record or block that claims more is refused before it is read. :raises CaptureFormatError: the input is not a capture, is cut short, or states a length over its limit. :raises OSError: the path cannot be opened or the stream cannot be read. :raises ValueError: max_frame_size is not positive.

Source code in src/pktcap/_container.py

def read_frames(
    source: CaptureSource, *, max_frame_size: int = _DEFAULT_MAX_FRAME_SIZE
) -> Iterator[CapturedFrame]:
    """Every packet of a pcap or pcapng capture, in file order.

    Nothing is read until the first frame is asked for, and frames are yielded
    as they are read, so a live pipe (``sys.stdin.buffer``) works and a damaged
    capture gives every frame before the damage and then raises. An empty
    input is an empty capture.

    :param source: a path, or a binary stream, which need not be seekable. A
        stream stays the caller's to close.
    :param max_frame_size: the most octets one captured frame may have. A
        record or block that claims more is refused before it is read.
    :raises CaptureFormatError: the input is not a capture, is cut short, or
        states a length over its limit.
    :raises OSError: the path cannot be opened or the stream cannot be read.
    :raises ValueError: ``max_frame_size`` is not positive.
    """
    if isinstance(max_frame_size, bool) or not isinstance(max_frame_size, int):
        raise TypeError("max_frame_size must be an int")
    if max_frame_size <= 0:
        raise ValueError("max_frame_size must be positive")
    if not isinstance(source, (str, os.PathLike)) and not hasattr(source, "read"):
        raise TypeError("source must be a path or a binary stream")
    return _open_and_read(source, max_frame_size)

command_hook(command, *, format='json', timeout=10.0, fail_fast=False, names=None, datagrams=False)

A hook that runs a program for each frame it is given.

Pass it to :func:copy_frames as each. The program is found once, here, as an absolute path: a name with a directory part is that file from the working directory then, a bare name is looked up on PATH. It is started with no argument and no shell, in its own process group, with the record of the frame (:func:frame_record, or :func:datagram_record of its datagram with datagrams) on standard input in format, a copy of the environment, PKTCAP_HOOK_FORMAT and one PKTCAP_HOOK_<FIELD> for each item names(frame) gives. A value passes the same rule as a field of a file name: no separator, no control character, at most 64 characters. A field name is the caller's code: letters, digits and underscores.

Past timeout seconds the program and everything it started are killed. A run that exits with a non-zero status or is killed is a failure: it is counted in the failures attribute of the returned callable and logged on pktcap._copy._hook at ERROR with a bounded, escaped tail of the error output, the first one and then at most one a minute. With fail_fast the first failure raises :class:CaptureHookError and so ends a copy.

:param command: a program name or path. :param format: the record format of standard input, one of :data:RECORD_FORMATS. :param timeout: seconds one run may take, positive and finite. :param names: fields for the environment, as for :func:copy_frames. :param datagrams: hand over the record of the frame's datagram. :raises ValueError: at the call, a command that does not exist, is not a file, is not executable, is a batch file on Windows or is empty; a timeout that is not positive and finite. On a call of the hook, a field name that is not a plain identifier or is set twice, or a frame with no datagram when datagrams is set. :raises TypeError: a command that is not text or a timeout that is not a number. :raises UnsupportedFormatError: format is no record format. :raises MissingExtraError: the format's extra is not installed.

Source code in src/pktcap/_copy/_hook.py

def command_hook(
    command: str,
    *,
    format: str = "json",
    timeout: float = 10.0,
    fail_fast: bool = False,
    names: Optional[Callable[[DissectedFrame], Mapping[str, object]]] = None,
    datagrams: bool = False,
) -> Callable[[DissectedFrame], None]:
    """A hook that runs a program for each frame it is given.

    Pass it to :func:`copy_frames` as ``each``. The program is found once,
    here, as an absolute path: a name with a directory part is that file from
    the working directory then, a bare name is looked up on ``PATH``. It is
    started with no argument and no shell, in its own process group, with the
    record of the frame (:func:`frame_record`, or :func:`datagram_record` of
    its datagram with ``datagrams``) on standard input in ``format``, a copy of
    the environment, ``PKTCAP_HOOK_FORMAT`` and one ``PKTCAP_HOOK_<FIELD>``
    for each item ``names(frame)`` gives. A value passes the same rule as a
    field of a file name: no separator, no control character, at most 64
    characters. A field name is the caller's code: letters, digits and
    underscores.

    Past ``timeout`` seconds the program and everything it started are killed.
    A run that exits with a non-zero status or is killed is a failure: it is
    counted in the ``failures`` attribute of the returned callable and logged
    on ``pktcap._copy._hook`` at ``ERROR`` with a bounded, escaped tail of the error
    output, the first one and then at most one a minute. With ``fail_fast``
    the first failure raises :class:`CaptureHookError` and so ends a copy.

    :param command: a program name or path.
    :param format: the record format of standard input, one of
        :data:`RECORD_FORMATS`.
    :param timeout: seconds one run may take, positive and finite.
    :param names: fields for the environment, as for :func:`copy_frames`.
    :param datagrams: hand over the record of the frame's datagram.
    :raises ValueError: at the call, a command that does not exist, is not a
        file, is not executable, is a batch file on Windows or is empty; a
        ``timeout`` that is not positive and finite. On a call of the hook, a
        field name that is not a plain identifier or is set twice, or a frame
        with no datagram when ``datagrams`` is set.
    :raises TypeError: a ``command`` that is not text or a ``timeout`` that is
        not a number.
    :raises UnsupportedFormatError: ``format`` is no record format.
    :raises MissingExtraError: the format's extra is not installed.
    """
    path = find_program(command)
    if isinstance(timeout, bool) or not isinstance(timeout, (int, float)):
        raise TypeError("timeout must be a number of seconds")
    if not (math.isfinite(timeout) and timeout > 0):
        raise ValueError("timeout must be above 0 seconds and finite")
    if names is not None and not callable(names):
        raise TypeError("names must be callable")
    record = record_format(format)
    record.require()
    return _CommandHook(path, record, float(timeout), bool(fail_fast), names, datagrams)

copy_frames(frames, writer, *, select=None, datagrams=False, limit=None, names=None, each=None)

Write the frames select accepts to writer, in order.

frames is any iterable of :class:DissectedFrame: a file through :func:read_dissected, a live source through :func:sniff_frames. The writer is the caller's: it is neither opened nor closed here.

:param select: a predicate over a frame, such as compile_capture_filter(text, frame_filter); None accepts all. :param datagrams: write each frame's UDP datagram (reassembled) and skip a frame that has none, instead of the frame. :param limit: stop once this many items are written; the source is not read past that. A refused item does not count. :param names: called with each frame select kept, before it is written; the mapping it returns is the writer's names, the values of its fields. It is given the frame even when datagrams is set, and may be called again for the same frame by a hook. :param each: called with the frame after its item is written, in order, and not for a frame select dropped, one with no datagram, or one the writer turned away. An exception from it ends the copy; the item stays written. :raises TypeError: writer is not a :class:CaptureWriter, select, names or each is not callable, limit is not an int, names gives something that is not a mapping, or frames yields something that is not a :class:DissectedFrame. :raises ValueError: limit is below zero. :raises OSError: the writer could not write. What was written before it stays written; the counts of the call are lost with the exception, so read writer.written for them.

Source code in src/pktcap/_copy/_frames.py

def copy_frames(
    frames: Iterable[DissectedFrame],
    writer: CaptureWriter,
    *,
    select: Optional[Callable[[DissectedFrame], bool]] = None,
    datagrams: bool = False,
    limit: Optional[int] = None,
    names: Optional[Callable[[DissectedFrame], Mapping[str, object]]] = None,
    each: Optional[Callable[[DissectedFrame], object]] = None,
) -> CopyResult:
    """Write the frames ``select`` accepts to ``writer``, in order.

    ``frames`` is any iterable of :class:`DissectedFrame`: a file through
    :func:`read_dissected`, a live source through :func:`sniff_frames`. The
    writer is the caller's: it is neither opened nor closed here.

    :param select: a predicate over a frame, such as
        ``compile_capture_filter(text, frame_filter)``; ``None`` accepts all.
    :param datagrams: write each frame's UDP datagram (reassembled) and skip
        a frame that has none, instead of the frame.
    :param limit: stop once this many items are written; the source is not
        read past that. A refused item does not count.
    :param names: called with each frame ``select`` kept, before it is
        written; the mapping it returns is the writer's ``names``, the values
        of its ``fields``. It is given the frame even when ``datagrams`` is
        set, and may be called again for the same frame by a hook.
    :param each: called with the frame after its item is written, in order,
        and not for a frame ``select`` dropped, one with no datagram, or one
        the writer turned away. An exception from it ends the copy; the item
        stays written.
    :raises TypeError: ``writer`` is not a :class:`CaptureWriter`, ``select``,
        ``names`` or ``each`` is not callable, ``limit`` is not an ``int``,
        ``names`` gives something that is not a mapping, or ``frames`` yields
        something that is not a :class:`DissectedFrame`.
    :raises ValueError: ``limit`` is below zero.
    :raises OSError: the writer could not write. What was written before it
        stays written; the counts of the call are lost with the exception, so
        read ``writer.written`` for them.
    """
    if not isinstance(writer, CaptureWriter):
        raise TypeError("writer must be a CaptureWriter")
    if select is not None and not callable(select):
        raise TypeError("select must be callable")
    if names is not None and not callable(names):
        raise TypeError("names must be callable")
    if each is not None and not callable(each):
        raise TypeError("each must be callable")
    if limit is not None:
        if isinstance(limit, bool) or not isinstance(limit, int):
            raise TypeError("limit must be an int")
        if limit < 0:
            raise ValueError("limit must not be below zero")
    source = iter(frames)
    read = written = skipped = refused = 0
    while limit is None or written < limit:
        try:
            frame = next(source)
        except StopIteration:
            break
        if not isinstance(frame, DissectedFrame):
            raise TypeError("frames must yield DissectedFrame values")
        read += 1
        if select is not None and not select(frame):
            skipped += 1
            continue
        item = frame.datagram() if datagrams else frame
        if item is None:
            skipped += 1
            continue
        values = None
        if names is not None:
            values = names(frame)
            if not isinstance(values, Mapping):
                raise TypeError("names must give a mapping")
        turned_away = writer.refused
        writer.write(item, names=values)
        if writer.refused > turned_away:
            refused += 1
        else:
            written += 1
            if each is not None:
                each(frame)
    return CopyResult(read, written, skipped, refused)

read_datagrams(source, *, dissector=None, max_frame_size=262144)

Every UDP datagram in a pcap or pcapng capture, in capture order.

:func:read_dissected through :meth:DissectedFrame.datagram: the view a UDP protocol library works on. IP fragments are reassembled. Frames that carry no UDP are not yielded; pass a dissector to read its :attr:~FrameDissector.stats and tell an empty capture from one of another link type.

:raises CaptureFormatError: the container is not a capture or is damaged.

Source code in src/pktcap/_dissect.py

def read_datagrams(
    source: CaptureSource,
    *,
    dissector: Optional[FrameDissector] = None,
    max_frame_size: int = 262144,
) -> Iterator[CapturedDatagram]:
    """Every UDP datagram in a pcap or pcapng capture, in capture order.

    :func:`read_dissected` through :meth:`DissectedFrame.datagram`: the view
    a UDP protocol library works on. IP fragments are reassembled. Frames
    that carry no UDP are not yielded; pass a ``dissector`` to read its
    :attr:`~FrameDissector.stats` and tell an empty capture from one of
    another link type.

    :raises CaptureFormatError: the container is not a capture or is damaged.
    """
    dissected = read_dissected(
        source, dissector=dissector, max_frame_size=max_frame_size
    )
    return _datagrams(dissected)

read_dissected(source, *, dissector=None, max_frame_size=262144)

Every frame of a pcap or pcapng capture, dissected, in capture order.

:func:read_frames and :meth:FrameDissector.dissect in one call. Every frame is yielded, whatever its link type and whatever it carries.

:param source: a path, or a binary stream, which need not be seekable. :param dissector: the frame dissector to use, for its registry, its options and its :attr:~FrameDissector.stats; a new FrameDissector() by default. :param max_frame_size: as for :func:read_frames. :raises CaptureFormatError: the container is not a capture or is damaged.

Source code in src/pktcap/_dissect.py

def read_dissected(
    source: CaptureSource,
    *,
    dissector: Optional[FrameDissector] = None,
    max_frame_size: int = 262144,
) -> Iterator[DissectedFrame]:
    """Every frame of a pcap or pcapng capture, dissected, in capture order.

    :func:`read_frames` and :meth:`FrameDissector.dissect` in one call. Every
    frame is yielded, whatever its link type and whatever it carries.

    :param source: a path, or a binary stream, which need not be seekable.
    :param dissector: the frame dissector to use, for its registry, its
        options and its :attr:`~FrameDissector.stats`; a new
        ``FrameDissector()`` by default.
    :param max_frame_size: as for :func:`read_frames`.
    :raises CaptureFormatError: the container is not a capture or is damaged.
    """
    frames = read_frames(source, max_frame_size=max_frame_size)
    if dissector is None:
        dissector = FrameDissector()
    elif not isinstance(dissector, FrameDissector):
        raise TypeError("dissector must be a FrameDissector")
    return map(dissector.dissect, frames)

check_dissector(dissector, samples, *, rounds=2000, seed=0)

Assert that dissector keeps the contract whatever it is given.

Each sample is tried as it is and then rounds times damaged (octets changed, the tail cut off). For every input the dissector must return a :class:Dissected whose payload is bytes and not longer than the input, or raise ValueError. The suite the built-in dissectors pass, for a protocol library to run on its own.

:raises AssertionError: the first input that broke the contract, as hex, and what came out of it. :raises ValueError: no sample was given.

Source code in src/pktcap/_dissectors/__init__.py

def check_dissector(
    dissector: Dissector,
    samples: Iterable[bytes],
    *,
    rounds: int = 2000,
    seed: int = 0,
) -> None:
    """Assert that ``dissector`` keeps the contract whatever it is given.

    Each sample is tried as it is and then ``rounds`` times damaged (octets
    changed, the tail cut off). For every input the dissector must return a
    :class:`Dissected` whose payload is ``bytes`` and not longer than the
    input, or raise ``ValueError``. The suite the built-in dissectors pass, for a
    protocol library to run on its own.

    :raises AssertionError: the first input that broke the contract, as hex,
        and what came out of it.
    :raises ValueError: no sample was given.
    """
    seeds = [bytes(sample) for sample in samples]
    if not seeds:
        raise ValueError("check_dissector needs at least one sample")
    chooser = random.Random(seed)
    inputs = list(seeds)
    for _ in range(rounds):
        damaged = bytearray(chooser.choice(seeds))
        for _ in range(chooser.randint(1, 3)):
            if damaged and chooser.random() < 0.7:
                damaged[chooser.randrange(len(damaged))] = chooser.randrange(256)
            else:
                del damaged[chooser.randrange(len(damaged) + 1) :]
        inputs.append(bytes(damaged))
    for data in inputs:
        try:
            result = dissector(data)
        except ValueError:
            continue
        except Exception as exc:
            raise AssertionError(
                "the dissector raised %s, not ValueError, for %s"
                % (type(exc).__name__, data.hex())
            ) from exc
        problem = _contract_problem(result, data)
        if problem:
            raise AssertionError("%s, for %s" % (problem, data.hex()))

default_registry()

The registry a :class:FrameDissector uses when given none.

One object for the process: what is registered into it is seen by every caller that did not pass a registry of its own.

Source code in src/pktcap/_dissectors/__init__.py

def default_registry() -> DissectorRegistry:
    """The registry a :class:`FrameDissector` uses when given none.

    One object for the process: what is registered into it is seen by every
    caller that did not pass a registry of its own.
    """
    return _DEFAULT

register_dissector(kind, value, dissector, *, replace=False)

Register dissector in :func:default_registry. See :meth:DissectorRegistry.register.

Source code in src/pktcap/_dissectors/__init__.py

def register_dissector(
    kind: str, value: int, dissector: Dissector, *, replace: bool = False
) -> None:
    """Register ``dissector`` in :func:`default_registry`. See
    :meth:`DissectorRegistry.register`."""
    _DEFAULT.register(kind, value, dissector, replace=replace)

compile_capture_filter(text, build)

One predicate for a capture-filter expression.

build is called once per clause, here and never per item, and returns the test for that clause's key and value; a != clause is inverted for it, and the predicate is true when every clause holds. So a key this library has never heard of, or a value that does not convert, is one error when the filter is compiled and not one per packet.

:param text: the expression; None or blank matches everything. :param build: turns one :class:FilterClause into a test. It should raise ValueError for an unknown key or a bad value. :raises CaptureFilterError: the expression is malformed, or build raised ValueError (which is chained, the clause named). :raises TypeError: build returned something that cannot be called.

Source code in src/pktcap/_filter.py

def compile_capture_filter(
    text: Optional[str], build: Callable[[FilterClause], Callable[[_T], bool]]
) -> Callable[[_T], bool]:
    """One predicate for a capture-filter expression.

    ``build`` is called once per clause, here and never per item, and returns
    the test for that clause's key and value; a ``!=`` clause is inverted for
    it, and the predicate is true when every clause holds. So a key this
    library has never heard of, or a value that does not convert, is one
    error when the filter is compiled and not one per packet.

    :param text: the expression; ``None`` or blank matches everything.
    :param build: turns one :class:`FilterClause` into a test. It should
        raise ``ValueError`` for an unknown key or a bad value.
    :raises CaptureFilterError: the expression is malformed, or ``build``
        raised ``ValueError`` (which is chained, the clause named).
    :raises TypeError: ``build`` returned something that cannot be called.
    """
    checks: List[Tuple[Callable[[_T], bool], bool]] = []
    for clause in parse_capture_filter(text):
        try:
            test = build(clause)
        except CaptureFilterError:
            raise
        except ValueError as exc:
            raise CaptureFilterError("%s: %s" % (clause, exc)) from exc
        if not callable(test):
            raise TypeError("build must return a callable for %s" % (clause,))
        checks.append((test, clause.negated))
    if not checks:
        return _everything

    def predicate(item: _T) -> bool:
        for test, negated in checks:
            if bool(test(item)) is negated:
                return False
        return True

    return predicate

parse_capture_filter(text)

The clauses of a capture-filter expression, in order.

:param text: the expression. None, empty or blank is no clause at all. :raises CaptureFilterError: a clause that is not key=value or key!=value, an or, an expression over 4,096 characters. :raises TypeError: text is neither str nor None.

Source code in src/pktcap/_filter.py

def parse_capture_filter(text: Optional[str]) -> Tuple[FilterClause, ...]:
    """The clauses of a capture-filter expression, in order.

    :param text: the expression. ``None``, empty or blank is no clause at all.
    :raises CaptureFilterError: a clause that is not ``key=value`` or
        ``key!=value``, an ``or``, an expression over 4,096 characters.
    :raises TypeError: ``text`` is neither ``str`` nor ``None``.
    """
    if text is None:
        return ()
    if not isinstance(text, str):
        raise TypeError("a capture filter is text")
    if len(text) > _MAX_FILTER_LENGTH:
        raise CaptureFilterError(
            "a capture filter of %d characters is over the limit of %d"
            % (len(text), _MAX_FILTER_LENGTH)
        )
    if not text.strip():
        return ()
    clauses: List[FilterClause] = []
    for part in _AND.split(text.strip()):
        clause = part.strip()
        key, operator, value = clause.partition("=")
        negated = key.endswith("!")
        key, value = key[: -1 if negated else None].strip(), value.strip()
        if _OR.search(clause) or key.lower() == "or":
            raise CaptureFilterError("capture filters join clauses with 'and' only")
        if not operator or not value or not _KEY.match(key):
            raise CaptureFilterError(
                "expected key=value or key!=value, got %r" % clause
            )
        clauses.append(FilterClause(key, value, negated))
    return tuple(clauses)

dumps_record(record, format='json')

One record as text in a named format, ending in a newline.

:param record: a mapping of text keys to plain data: dict, list, str, int, float, bool, None. :param format: one of :data:RECORD_FORMATS. :raises UnsupportedFormatError: no record format of that name. :raises MissingExtraError: the format's extra is not installed; it is an ImportError and its message names the extra. :raises TypeError: record is not a mapping, or holds a value the format cannot represent. :raises ValueError: a value the format must refuse, such as a NaN.

Source code in src/pktcap/_formats/__init__.py

def dumps_record(record: Mapping[str, Any], format: str = "json") -> str:
    """One record as text in a named format, ending in a newline.

    :param record: a mapping of text keys to plain data: ``dict``, ``list``,
        ``str``, ``int``, ``float``, ``bool``, ``None``.
    :param format: one of :data:`RECORD_FORMATS`.
    :raises UnsupportedFormatError: no record format of that name.
    :raises MissingExtraError: the format's extra is not installed; it is an
        ``ImportError`` and its message names the extra.
    :raises TypeError: ``record`` is not a mapping, or holds a value the
        format cannot represent.
    :raises ValueError: a value the format must refuse, such as a NaN.
    """
    fmt = record_format(format)
    if not isinstance(record, Mapping):
        raise TypeError("a record is a mapping")
    return fmt.dumps(record)

has_output_format(name)

Whether the format of that name can be written on this installation.

False for a format whose extra is not installed; the name stays in :data:OUTPUT_FORMATS either way. An unknown name raises UnsupportedFormatError.

Source code in src/pktcap/_formats/__init__.py

def has_output_format(name: str) -> bool:
    """Whether the format of that name can be written on this installation.

    ``False`` for a format whose extra is not installed; the name stays in
    :data:`OUTPUT_FORMATS` either way. An unknown name raises
    ``UnsupportedFormatError``.
    """
    lowered = _normalised(name, OUTPUT_FORMATS)
    if lowered in CAPTURE_FORMATS or lowered == TEXT_FORMAT:
        return True
    try:
        _RECORD_FORMATS[lowered].require()
    except ImportError:
        return False
    return True

loads_record(text, format='json')

The one record text holds, as written by :func:dumps_record.

:param text: the content (never a file name) of one record. :param format: one of :data:RECORD_FORMATS. :raises UnsupportedFormatError: no record format of that name. :raises MissingExtraError: the format cannot be read on this installation (yaml without PyYAML; toml on Python before 3.11 without tomli); it is an ImportError and its message names the extra. :raises TypeError: text is not a str. :raises RecordFormatError: text is not one record in that format. The message says what is wrong and where and never quotes the text.

Source code in src/pktcap/_formats/__init__.py

def loads_record(text: str, format: str = "json") -> Dict[str, Any]:
    """The one record ``text`` holds, as written by :func:`dumps_record`.

    :param text: the content (never a file name) of one record.
    :param format: one of :data:`RECORD_FORMATS`.
    :raises UnsupportedFormatError: no record format of that name.
    :raises MissingExtraError: the format cannot be read on this installation
        (``yaml`` without PyYAML; ``toml`` on Python before 3.11 without
        ``tomli``); it is an ``ImportError`` and its message names the extra.
    :raises TypeError: ``text`` is not a ``str``.
    :raises RecordFormatError: ``text`` is not one record in that format. The
        message says what is wrong and where and never quotes the text.
    """
    fmt = record_format(format)
    if not isinstance(text, str):
        raise TypeError("the text of a record is a str")
    return fmt.loads(text)

frame_filter(clause)

The test of one filter clause against a :class:DissectedFrame.

A comma in the value means "any of". The keys:

src, dst, host the source, the destination, or either address of the outermost IP layer: an address or a network (10.0.0.5, 10.0.0.0/8, 2001:db8::/32). sport, dport, port the source, the destination, or either port of the UDP or TCP layer. proto a layer the frame has, by name (udp, tcp, ipv4, ipv6, vlan, ethernet, or the name of a registered dissector's layer: its class name in lower case, without a trailing Layer), or an IP protocol number (17). vlan a VLAN identifier on any tag of the frame. linktype the capture's link-type number. LAYER.FIELD a field of a built-in layer (ipv4.ttl, udp.destination_port), compared by the type of its value.

:raises ValueError: a key that is none of these, or a value that is not what the key takes. :func:compile_capture_filter turns it into a CaptureFilterError naming the clause.

Source code in src/pktcap/_frame_filter.py

def frame_filter(clause: FilterClause) -> Predicate:
    """The test of one filter clause against a :class:`DissectedFrame`.

    A comma in the value means "any of". The keys:

    ``src``, ``dst``, ``host``
        the source, the destination, or either address of the outermost IP
        layer: an address or a network (``10.0.0.5``, ``10.0.0.0/8``,
        ``2001:db8::/32``).
    ``sport``, ``dport``, ``port``
        the source, the destination, or either port of the UDP or TCP layer.
    ``proto``
        a layer the frame has, by name (``udp``, ``tcp``, ``ipv4``, ``ipv6``,
        ``vlan``, ``ethernet``, or the name of a registered dissector's layer:
        its class name in lower case, without a trailing ``Layer``), or an IP
        protocol number (``17``).
    ``vlan``
        a VLAN identifier on any tag of the frame.
    ``linktype``
        the capture's link-type number.
    ``LAYER.FIELD``
        a field of a built-in layer (``ipv4.ttl``, ``udp.destination_port``),
        compared by the type of its value.

    :raises ValueError: a key that is none of these, or a value that is not
        what the key takes. :func:`compile_capture_filter` turns it into a
        ``CaptureFilterError`` naming the clause.
    """
    return _answer(clause, None, Guard())

frame_filter_for(registry)

The build of :func:compile_capture_filter for a registry: the keys of :func:frame_filter, plus LAYER.KEY and LAYER.FIELD of every layer the registry declares, and a registered key by its bare name while exactly one layer has it.

:raises TypeError: registry is not a :class:DissectorRegistry.

Source code in src/pktcap/_frame_filter.py

def frame_filter_for(
    registry: DissectorRegistry,
) -> Callable[[FilterClause], Predicate]:
    """The ``build`` of :func:`compile_capture_filter` for a registry: the
    keys of :func:`frame_filter`, plus ``LAYER.KEY`` and ``LAYER.FIELD`` of
    every layer the registry declares, and a registered key by its bare name
    while exactly one layer has it.

    :raises TypeError: ``registry`` is not a :class:`DissectorRegistry`.
    """
    if not isinstance(registry, DissectorRegistry):
        raise TypeError("frame_filter_for takes a DissectorRegistry")
    guard = Guard()

    def build(clause: FilterClause) -> Predicate:
        return _answer(clause, registry, guard)

    return build

frame_filter_keys(registry=None)

Every filter key that compiles, sorted: the built-in ones, each LAYER.KEY and LAYER.FIELD of the built-in layers and of the registry's, and the bare form of a registered key exactly one layer has.

:raises TypeError: registry is neither None nor a registry.

Source code in src/pktcap/_frame_filter.py

def frame_filter_keys(registry: Optional[DissectorRegistry] = None) -> Tuple[str, ...]:
    """Every filter key that compiles, sorted: the built-in ones, each
    ``LAYER.KEY`` and ``LAYER.FIELD`` of the built-in layers and of the
    registry's, and the bare form of a registered key exactly one layer has.

    :raises TypeError: ``registry`` is neither ``None`` nor a registry.
    """
    if registry is not None and not isinstance(registry, DissectorRegistry):
        raise TypeError("frame_filter_keys takes a DissectorRegistry or None")
    return list_keys(registry, FRAME_FILTER_KEYS)

has_live_capture()

Whether this platform can capture live: whether it has AF_PACKET.

It says nothing about permission: opening the socket still needs CAP_NET_RAW.

Source code in src/pktcap/_live.py

def has_live_capture() -> bool:
    """Whether this platform can capture live: whether it has ``AF_PACKET``.

    It says nothing about permission: opening the socket still needs
    ``CAP_NET_RAW``.
    """
    return _AF_PACKET is not None

sniff(interface=None, *, stop=None, dissector=None)

UDP datagrams seen on interface, as they arrive. Linux only.

:func:sniff_frames through the datagram view, for a UDP protocol library; for every packet, use that or iterate a :class:LiveCapture. The socket is opened when the first datagram is asked for and closed when the iterator ends or is closed.

:param interface: as for :class:LiveCapture; None is every one. :param stop: called between packets, and at least once a second on a quiet interface; returning true ends the iteration. :param dissector: the frame dissector to use; a new FrameDissector() by default. :raises LiveCaptureError: when the first datagram is asked for, where the platform cannot capture; ask :func:has_live_capture beforehand. :raises PermissionError: at the same point, without CAP_NET_RAW.

Source code in src/pktcap/_live.py

def sniff(
    interface: InterfaceLike = None,
    *,
    stop: Optional[Callable[[], bool]] = None,
    dissector: Optional[FrameDissector] = None,
) -> Iterator[CapturedDatagram]:
    """UDP datagrams seen on ``interface``, as they arrive. Linux only.

    :func:`sniff_frames` through the datagram view, for a UDP protocol
    library; for every packet, use that or iterate a :class:`LiveCapture`. The
    socket is opened when the first datagram is asked for and closed when the
    iterator ends or is closed.

    :param interface: as for :class:`LiveCapture`; ``None`` is every one.
    :param stop: called between packets, and at least once a second on a
        quiet interface; returning true ends the iteration.
    :param dissector: the frame dissector to use; a new ``FrameDissector()``
        by default.
    :raises LiveCaptureError: when the first datagram is asked for, where the
        platform cannot capture; ask :func:`has_live_capture` beforehand.
    :raises PermissionError: at the same point, without ``CAP_NET_RAW``.
    """
    frames = _sniff(LiveCapture(interface), stop, _checked(stop, dissector))
    return _datagrams(frames)

sniff_frames(interface=None, *, stop=None, dissector=None)

Every frame seen on interface, dissected, as they arrive. Linux only.

:class:LiveCapture and a :class:FrameDissector in one call: what :func:read_dissected is for a file. The socket is opened when the first frame is asked for and closed when the iterator ends or is closed.

:param interface: as for :class:LiveCapture; None is every one. :param stop: called between packets, and at least once a second on a quiet interface; returning true ends the iteration. :param dissector: the frame dissector to use; a new FrameDissector() by default. :raises LiveCaptureError: when the first frame is asked for, where the platform cannot capture; ask :func:has_live_capture beforehand. :raises PermissionError: at the same point, without CAP_NET_RAW. :raises TypeError: stop is not callable or dissector is not a :class:FrameDissector, at the call.

Source code in src/pktcap/_live.py

def sniff_frames(
    interface: InterfaceLike = None,
    *,
    stop: Optional[Callable[[], bool]] = None,
    dissector: Optional[FrameDissector] = None,
) -> Iterator[DissectedFrame]:
    """Every frame seen on ``interface``, dissected, as they arrive. Linux only.

    :class:`LiveCapture` and a :class:`FrameDissector` in one call: what
    :func:`read_dissected` is for a file. The socket is opened when the first
    frame is asked for and closed when the iterator ends or is closed.

    :param interface: as for :class:`LiveCapture`; ``None`` is every one.
    :param stop: called between packets, and at least once a second on a
        quiet interface; returning true ends the iteration.
    :param dissector: the frame dissector to use; a new ``FrameDissector()``
        by default.
    :raises LiveCaptureError: when the first frame is asked for, where the
        platform cannot capture; ask :func:`has_live_capture` beforehand.
    :raises PermissionError: at the same point, without ``CAP_NET_RAW``.
    :raises TypeError: ``stop`` is not callable or ``dissector`` is not a
        :class:`FrameDissector`, at the call.
    """
    return _sniff(LiveCapture(interface), stop, _checked(stop, dissector))

load_plugins(registry, plugins=None, *, always=())

Import the plugins a list names and call each one's hook with registry.

The list is plugins and nothing else: this reads no variable and no file. An item is a dotted Python name: a module with a pktcap_plugin(registry) function, or MODULE.CALLABLE for any callable taking the registry. Items are separated by , ; : or white space; none alone is the empty list. Importing a module runs its code, so a program passes a list only from a place it trusts.

:param registry: where the hooks register. There is no default one. :param plugins: None (no plugin), one text, or an iterable of texts. :param always: items a caller's code loads whatever the user listed, as dotted names written as for plugins. They load first, with the source "always". An item of plugins that resolves to the same hook callable is skipped, not loaded twice and not an error; a different callable that claims a selector already taken is the plugin error. :returns: one :class:LoadedPlugin per item loaded, in order. :raises CapturePluginError: an item cannot be loaded. The registry is left as it was before the call; modules already imported stay imported. :raises ValueError: a malformed item of plugins or always. :raises TypeError: registry is not a :class:DissectorRegistry, or a list is not text.

Source code in src/pktcap/_plugins/_load.py

def load_plugins(
    registry: DissectorRegistry,
    plugins: Union[None, str, Iterable[str]] = None,
    *,
    always: Union[str, Iterable[str]] = (),
) -> Tuple[LoadedPlugin, ...]:
    """Import the plugins a list names and call each one's hook with ``registry``.

    The list is ``plugins`` and nothing else: this reads no variable and no
    file. An item is a dotted Python name: a module with a
    ``pktcap_plugin(registry)`` function, or ``MODULE.CALLABLE`` for any
    callable taking the registry. Items are separated by ``,`` ``;`` ``:`` or
    white space; ``none`` alone is the empty list. Importing a module runs its
    code, so a program passes a list only from a place it trusts.

    :param registry: where the hooks register. There is no default one.
    :param plugins: ``None`` (no plugin), one text, or an iterable of texts.
    :param always: items a caller's code loads whatever the user listed, as
        dotted names written as for ``plugins``. They load first, with the
        source ``"always"``. An item of ``plugins`` that resolves to the same
        hook callable is skipped, not loaded twice and not an error; a
        different callable that claims a selector already taken is the plugin
        error.
    :returns: one :class:`LoadedPlugin` per item loaded, in order.
    :raises CapturePluginError: an item cannot be loaded. The registry is left
        as it was before the call; modules already imported stay imported.
    :raises ValueError: a malformed item of ``plugins`` or ``always``.
    :raises TypeError: ``registry`` is not a :class:`DissectorRegistry`, or a
        list is not text.
    """
    return load_named(registry, plugins, PLUGINS_ARGUMENT, always)

datagram_record(datagram)

The record of a datagram for a caller with no protocol to decode it: time, source, destination, length and payload as hex, plus fragmented or truncated when the datagram is partial.

source and destination are host:port text, an IPv6 host in brackets. time is the number of seconds the capture states.

Source code in src/pktcap/_records.py

def datagram_record(datagram: CapturedDatagram) -> Dict[str, Any]:
    """The record of a datagram for a caller with no protocol to decode it:
    ``time``, ``source``, ``destination``, ``length`` and ``payload`` as hex,
    plus ``fragmented`` or ``truncated`` when the datagram is partial.

    ``source`` and ``destination`` are ``host:port`` text, an IPv6 host in
    brackets. ``time`` is the number of seconds the capture states.
    """
    record: Dict[str, Any] = {
        "time": datagram.time,
        "source": join_host(datagram.source[0], datagram.source[1]),
        "destination": join_host(datagram.destination[0], datagram.destination[1]),
        "length": len(datagram.payload),
        "payload": datagram.payload.hex(),
    }
    if datagram.fragmented:
        record["fragmented"] = True
    if datagram.truncated:
        record["truncated"] = True
    return record

frame_record(frame)

The record of a dissected frame: time, linktype, length (of the captured frame), layers and payload.

layers is a list, outermost first, of one mapping per layer: layer is its name (ethernet, ipv4, udp: the class name without Layer, in lower case) and the rest are its fields, octets as hex. payload is the hex of what no dissector read. interface, error and reassembled appear only when they say something.

Source code in src/pktcap/_records.py

def frame_record(frame: DissectedFrame) -> Dict[str, Any]:
    """The record of a dissected frame: ``time``, ``linktype``, ``length``
    (of the captured frame), ``layers`` and ``payload``.

    ``layers`` is a list, outermost first, of one mapping per layer: ``layer``
    is its name (``ethernet``, ``ipv4``, ``udp``: the class name without
    ``Layer``, in lower case) and the rest are its fields, octets as hex.
    ``payload`` is the hex of what no dissector read. ``interface``,
    ``error`` and ``reassembled`` appear only when they say something.
    """
    record: Dict[str, Any] = {
        "time": frame.frame.time,
        "linktype": frame.frame.linktype,
        "length": len(frame.frame.data),
        "layers": [_layer_record(layer) for layer in frame.layers],
        "payload": frame.payload.hex(),
    }
    if frame.frame.interface is not None:
        record["interface"] = frame.frame.interface
    if frame.error is not None:
        record["error"] = frame.error
    if frame.reassembled:
        record["reassembled"] = True
    return record

replay(source, deliver, *, speed=1.0, max_delay=5.0, limit=None)

replay(
    source: CaptureSource,
    deliver: Callable[[CapturedDatagram], object],
    *,
    speed: Optional[float] = ...,
    max_delay: float = ...,
    limit: Optional[int] = ...
) -> int

replay(
    source: Iterable[_T],
    deliver: Callable[[_T], object],
    *,
    speed: Optional[float] = ...,
    max_delay: float = ...,
    limit: Optional[int] = ...
) -> int

Hand each item of a capture to deliver, in order and in time.

Blocks, sleeping between items as :func:replay_schedule says. What deliver does with an item is the caller's: a protocol library that knows a reply goes to another port builds a faithful replay on this, and frames that are not UDP are replayed by passing read_frames(path) or read_dissected(path) as the source. An exception from deliver ends the replay and propagates.

:param deliver: called with each item, partial datagrams included; what it returns is ignored. :returns: how many items were delivered.

Source code in src/pktcap/_replay.py

def replay(
    source: object,
    deliver: Callable[[Any], object],
    *,
    speed: Optional[float] = 1.0,
    max_delay: float = 5.0,
    limit: Optional[int] = None,
) -> int:
    """Hand each item of a capture to ``deliver``, in order and in time.

    Blocks, sleeping between items as :func:`replay_schedule` says. What
    ``deliver`` does with an item is the caller's: a protocol library that
    knows a reply goes to another port builds a faithful replay on this, and
    frames that are not UDP are replayed by passing ``read_frames(path)`` or
    ``read_dissected(path)`` as the source. An exception from ``deliver`` ends
    the replay and propagates.

    :param deliver: called with each item, partial datagrams included; what
        it returns is ignored.
    :returns: how many items were delivered.
    """
    if not callable(deliver):
        raise TypeError("deliver must be callable")
    delivered = 0
    for delay, item in _paced(source, speed, max_delay, limit):
        if delay:
            time.sleep(delay)
        deliver(item)
        delivered += 1
    return delivered

replay_schedule(source, *, speed=1.0, max_delay=5.0, limit=None)

replay_schedule(
    source: CaptureSource,
    *,
    speed: Optional[float] = ...,
    max_delay: float = ...,
    limit: Optional[int] = ...
) -> Iterator[Tuple[float, CapturedDatagram]]

replay_schedule(
    source: Iterable[_T],
    *,
    speed: Optional[float] = ...,
    max_delay: float = ...,
    limit: Optional[int] = ...
) -> Iterator[Tuple[float, _T]]

(delay, item) for each item of a capture, in order.

delay is how many seconds to wait before that item: the time since the one before it in the capture, divided by speed and never more than max_delay. This reads no clock and waits for nothing, so an event loop can drive it with its own sleep.

:param source: a path or binary stream of a pcap or pcapng capture, which is read as UDP datagrams; or an iterable of anything with a time in seconds, such as read_frames(path) for every frame or a generator that selects the datagrams to replay. :param speed: 1.0 keeps the recorded timing, 2.0 halves every wait; None makes every delay zero. :param max_delay: the longest single wait, in seconds. The capture controls its timestamps: a jump of a year must not hang a replay. A step backwards in time is a wait of zero. :param limit: stop after this many items. :raises ValueError: a speed that is not positive, a negative max_delay or limit. :raises CaptureFormatError: while iterating, for a damaged capture.

Source code in src/pktcap/_replay.py

def replay_schedule(
    source: object,
    *,
    speed: Optional[float] = 1.0,
    max_delay: float = 5.0,
    limit: Optional[int] = None,
) -> Iterator[Tuple[float, Any]]:
    """``(delay, item)`` for each item of a capture, in order.

    ``delay`` is how many seconds to wait before that item: the time since the
    one before it in the capture, divided by ``speed`` and never more than
    ``max_delay``. This reads no clock and waits for nothing, so an event loop
    can drive it with its own sleep.

    :param source: a path or binary stream of a pcap or pcapng capture, which
        is read as UDP datagrams; or an iterable of anything with a ``time``
        in seconds, such as ``read_frames(path)`` for every frame or a
        generator that selects the datagrams to replay.
    :param speed: ``1.0`` keeps the recorded timing, ``2.0`` halves every
        wait; ``None`` makes every delay zero.
    :param max_delay: the longest single wait, in seconds. The capture
        controls its timestamps: a jump of a year must not hang a replay. A
        step backwards in time is a wait of zero.
    :param limit: stop after this many items.
    :raises ValueError: a ``speed`` that is not positive, a negative
        ``max_delay`` or ``limit``.
    :raises CaptureFormatError: while iterating, for a damaged capture.
    """
    return _paced(source, speed, max_delay, limit)

replay_to(source, dst, port, *, endpoint=None, speed=1.0, max_delay=5.0, limit=None)

Send the payload of each UDP datagram of a capture to (dst, port).

Blocks. Every datagram goes to the one destination named here; the addresses recorded in the capture are never sent to. A datagram marked fragmented or truncated is counted and not sent.

:param source: a capture, or an iterable of :class:CapturedDatagram. :param dst: where to send: an address or a host name, looked up once. :param port: the UDP port there, 1 to 65535. :param endpoint: a :class:netimps.UDPEndpoint to send through, left open. By default a socket is bound for dst's address family with netimps.bind's defaults (any free port, no broadcast) and closed afterwards: pass an endpoint to choose the source port or interface, or to send to a broadcast address. :param speed: as for :func:replay_schedule; the recorded timing by default, so a replay sends no faster than the capture did. :raises OSError: dst does not resolve, or a send fails; a payload longer than the destination's family allows is one of these. :raises ValueError: a port outside 1-65535, or a bad option. :raises TypeError: an item of source is not a datagram.

Source code in src/pktcap/_replay.py

def replay_to(
    source: Union[CaptureSource, Iterable[CapturedDatagram]],
    dst: HostLike,
    port: int,
    *,
    endpoint: Optional[UDPEndpoint] = None,
    speed: Optional[float] = 1.0,
    max_delay: float = 5.0,
    limit: Optional[int] = None,
) -> ReplayResult:
    """Send the payload of each UDP datagram of a capture to ``(dst, port)``.

    Blocks. Every datagram goes to the one destination named here; the
    addresses recorded in the capture are never sent to. A datagram marked
    ``fragmented`` or ``truncated`` is counted and not sent.

    :param source: a capture, or an iterable of :class:`CapturedDatagram`.
    :param dst: where to send: an address or a host name, looked up once.
    :param port: the UDP port there, 1 to 65535.
    :param endpoint: a :class:`netimps.UDPEndpoint` to send through, left
        open. By default a socket is bound for ``dst``'s address family with
        ``netimps.bind``'s defaults (any free port, no broadcast) and closed
        afterwards: pass an endpoint to choose the source port or interface,
        or to send to a broadcast address.
    :param speed: as for :func:`replay_schedule`; the recorded timing by
        default, so a replay sends no faster than the capture did.
    :raises OSError: ``dst`` does not resolve, or a send fails; a payload
        longer than the destination's family allows is one of these.
    :raises ValueError: a ``port`` outside 1-65535, or a bad option.
    :raises TypeError: an item of ``source`` is not a datagram.
    """
    if isinstance(port, bool) or not isinstance(port, int):
        raise TypeError("port must be an int")
    if not 1 <= port <= 65535:
        raise ValueError("port %d is outside 1-65535" % port)
    if dst is None or (isinstance(dst, str) and not dst.strip()):
        raise ValueError("dst must name the host to send to")
    schedule = _paced(source, speed, max_delay, limit)
    address = Host(dst).ip(check=True)
    if address is None:  # check=True raises for a name with no address
        raise OSError("%r has no address" % (dst,))
    own = None
    if endpoint is None:
        own = endpoint = UDPEndpoint(
            bind("::" if address.version == 6 else "", 0), pktinfo=False
        )
    sent = partial = 0
    try:
        for delay, datagram in schedule:
            if not isinstance(datagram, CapturedDatagram):
                raise TypeError("replay_to sends CapturedDatagram items only")
            if delay:
                time.sleep(delay)
            if datagram.fragmented or datagram.truncated:
                partial += 1
                continue
            endpoint.send(datagram.payload, str(address), port)
            sent += 1
    finally:
        if own is not None:
            own.close()
    return ReplayResult(sent, partial)

read_tcp_streams(source, *, reassembler=None, dissector=None, max_frame_size=262144)

The TCP streams of a pcap or pcapng capture, in the order they become deliverable.

:func:read_dissected through :meth:TCPReassembler.add, then :meth:TCPReassembler.flush when the capture ends. Nothing is read before the first item is asked for, and a caller that stops early leaves the rest unread. A damaged container gives every item the frames before the damage made deliverable and what was still held, then raises.

:param source: a path, or a binary stream, which need not be seekable. :param reassembler: the reassembler to use, for its options and its :attr:~TCPReassembler.stats; a new TCPReassembler() by default. :param dissector: the frame dissector to use, as for :func:read_dissected. :param max_frame_size: as for :func:read_frames. :raises TypeError: a reassembler or dissector of the wrong type, before anything is read. :raises CaptureFormatError: the container is not a capture or is damaged.

Source code in src/pktcap/_streams/_read.py

def read_tcp_streams(
    source: CaptureSource,
    *,
    reassembler: Optional[TCPReassembler] = None,
    dissector: Optional[FrameDissector] = None,
    max_frame_size: int = 262144,
) -> Iterator[TCPStreamData]:
    """The TCP streams of a pcap or pcapng capture, in the order they become
    deliverable.

    :func:`read_dissected` through :meth:`TCPReassembler.add`, then
    :meth:`TCPReassembler.flush` when the capture ends. Nothing is read
    before the first item is asked for, and a caller that stops early leaves
    the rest unread. A damaged container gives every item the frames before
    the damage made deliverable and what was still held, then raises.

    :param source: a path, or a binary stream, which need not be seekable.
    :param reassembler: the reassembler to use, for its options and its
        :attr:`~TCPReassembler.stats`; a new ``TCPReassembler()`` by default.
    :param dissector: the frame dissector to use, as for
        :func:`read_dissected`.
    :param max_frame_size: as for :func:`read_frames`.
    :raises TypeError: a ``reassembler`` or ``dissector`` of the wrong type,
        before anything is read.
    :raises CaptureFormatError: the container is not a capture or is damaged.
    """
    if reassembler is None:
        reassembler = TCPReassembler()
    elif not isinstance(reassembler, TCPReassembler):
        raise TypeError("reassembler must be a TCPReassembler")
    frames = read_dissected(source, dissector=dissector, max_frame_size=max_frame_size)
    return _streams(frames, reassembler)

asniff_udp(endpoints, *, dissector=None)

:func:sniff_udp for an event loop, on UDPEndpoint.arecv.

One task for each endpoint feeds one bounded queue. When the iterator ends, fails, is cancelled or its aclose() is awaited, every task has been cancelled and every socket closed; aclose() is what a loop that stops early (break) awaits. Python imports asyncio only when the first frame is asked for.

:raises TypeError: dissector is not a :class:FrameDissector, or an item of endpoints is not a :class:netimps.UDPEndpoint. :raises ValueError: no endpoint, or more than 256. :raises OSError: a socket failed, while iterating.

Source code in src/pktcap/_sources/_sniff.py

def asniff_udp(
    endpoints: Iterable[UDPEndpoint], *, dissector: Optional[FrameDissector] = None
) -> AsyncIterator[DissectedFrame]:
    """:func:`sniff_udp` for an event loop, on ``UDPEndpoint.arecv``.

    One task for each endpoint feeds one bounded queue. When the iterator
    ends, fails, is cancelled or its ``aclose()`` is awaited, every task has
    been cancelled and every socket closed; ``aclose()`` is what a loop that
    stops early (``break``) awaits. Python imports ``asyncio`` only when the
    first frame is asked for.

    :raises TypeError: ``dissector`` is not a :class:`FrameDissector`, or an
        item of ``endpoints`` is not a :class:`netimps.UDPEndpoint`.
    :raises ValueError: no endpoint, or more than 256.
    :raises OSError: a socket failed, while iterating.
    """
    checked = _checked(None, dissector)
    return _AsyncFrames(UDPCapture(endpoints), checked)

sniff_udp(endpoints, *, stop=None, dissector=None)

Every datagram that reaches the endpoints, as a dissected raw-IP frame.

:class:UDPCapture and a :class:FrameDissector in one call, as :func:sniff_frames is for an interface. The endpoints are the iterator's from the call: it closes them when it ends, fails, is closed or is dropped. stop() is called between datagrams, and at least once a second on a quiet socket; returning true ends the iteration.

:raises TypeError: stop is not callable, dissector is not a :class:FrameDissector, or an item of endpoints is not a :class:netimps.UDPEndpoint; the endpoints are then still the caller's. :raises ValueError: no endpoint, or more than 256. :raises OSError: a socket failed, while iterating.

Source code in src/pktcap/_sources/_sniff.py

def sniff_udp(
    endpoints: Iterable[UDPEndpoint],
    *,
    stop: Optional[Callable[[], bool]] = None,
    dissector: Optional[FrameDissector] = None,
) -> Iterator[DissectedFrame]:
    """Every datagram that reaches the endpoints, as a dissected raw-IP frame.

    :class:`UDPCapture` and a :class:`FrameDissector` in one call, as
    :func:`sniff_frames` is for an interface. The endpoints are the iterator's
    from the call: it closes them when it ends, fails, is closed or is
    dropped. ``stop()`` is called between datagrams, and at least once a
    second on a quiet socket; returning true ends the iteration.

    :raises TypeError: ``stop`` is not callable, ``dissector`` is not a
        :class:`FrameDissector`, or an item of ``endpoints`` is not a
        :class:`netimps.UDPEndpoint`; the endpoints are then still the caller's.
    :raises ValueError: no endpoint, or more than 256.
    :raises OSError: a socket failed, while iterating.
    """
    checked = _checked(stop, dissector)
    return _Frames(UDPCapture(endpoints), stop, checked)

frame_summary(frame)

One line for a frame: its time (UTC), its two socket addresses when it has them, the name of its innermost layer and that layer's summary().

A layer with no summary method, or whose summary() raises or gives something that is not text, is described by its name; the failure is logged once for the layer's class. Everything is escaped to printable ASCII on one line and cut at 512 characters (... ends a cut line).

:raises TypeError: frame is not a :class:DissectedFrame.

Source code in src/pktcap/_summary.py

def frame_summary(frame: DissectedFrame) -> str:
    """One line for a frame: its time (UTC), its two socket addresses when it
    has them, the name of its innermost layer and that layer's ``summary()``.

    A layer with no ``summary`` method, or whose ``summary()`` raises or gives
    something that is not text, is described by its name; the failure is
    logged once for the layer's class. Everything is escaped to printable
    ASCII on one line and cut at 512 characters (``...`` ends a cut line).

    :raises TypeError: ``frame`` is not a :class:`DissectedFrame`.
    """
    if not isinstance(frame, DissectedFrame):
        raise TypeError("frame must be a DissectedFrame")
    parts: Tuple[str, ...] = (_time_text(frame.frame.time), _addresses(frame))
    if frame.layers:
        layer = frame.layers[-1]
        name, text = layer_name(layer), _summary_of(layer)
        parts += ("%s: %s" % (name, text) if text else name,)
    else:
        parts += (
            "linktype %d, %d octets" % (frame.frame.linktype, len(frame.frame.data)),
        )
    return _line(" ".join(part for part in parts if part))

datagram_frame(datagram, *, interface=None, ident=0)

A datagram seen at a socket as the raw-IP frame (link type 101) a dissector reads: the octets the writers put in a capture for it.

The IP header is made up, as it is in a capture file: version, TTL 64, no fragmentation, ident as the identification, valid checksums. A v4-mapped pair of addresses is an IPv4 packet, a pair of one IPv4 and one IPv6 address an IPv6 packet, and a zone is not on the wire. Dissected, the frame gives datagram back with those hosts.

:param interface: the number of the interface it arrived on, or None. :param ident: the IPv4 identification, kept to 16 bits. :raises ValueError: a host that is not an address, a port outside 0-65535, a payload over 65,507 octets (IPv4) or 65,527 (IPv6), a time out of range, or an interface outside 0 to 2**32 - 1. :raises TypeError: an argument of the wrong type.

Source code in src/pktcap/_writer.py

def datagram_frame(
    datagram: CapturedDatagram, *, interface: Optional[int] = None, ident: int = 0
) -> CapturedFrame:
    """A datagram seen at a socket as the raw-IP frame (link type 101) a
    dissector reads: the octets the writers put in a capture for it.

    The IP header is made up, as it is in a capture file: version, TTL 64, no
    fragmentation, ``ident`` as the identification, valid checksums. A v4-mapped
    pair of addresses is an IPv4 packet, a pair of one IPv4 and one IPv6 address
    an IPv6 packet, and a zone is not on the wire. Dissected, the frame gives
    ``datagram`` back with those hosts.

    :param interface: the number of the interface it arrived on, or ``None``.
    :param ident: the IPv4 identification, kept to 16 bits.
    :raises ValueError: a host that is not an address, a port outside 0-65535,
        a payload over 65,507 octets (IPv4) or 65,527 (IPv6), a time out of
        range, or an ``interface`` outside 0 to 2**32 - 1.
    :raises TypeError: an argument of the wrong type.
    """
    if not isinstance(datagram, CapturedDatagram):
        raise TypeError("datagram must be a CapturedDatagram")
    if interface is not None:
        if isinstance(interface, bool) or not isinstance(interface, int):
            raise TypeError("interface must be an int or None")
        if not 0 <= interface < _MAX_SECONDS:
            raise ValueError("interface %d is outside 0 to 2**32 - 1" % interface)
    if isinstance(ident, bool) or not isinstance(ident, int):
        raise TypeError("ident must be an int")
    _stamp(datagram.time)
    packet = _ip_packet(datagram.source, datagram.destination, datagram.payload, ident)
    return CapturedFrame(datagram.time, _LINKTYPE_RAW, packet, interface)