Base Class

pcapkit.foundation.traceflow.traceflow contains TraceFlow only, which is an abstract base class for all flow tracing classes.

class pcapkit.foundation.traceflow.traceflow.TraceFlow(fout, format, byteorder='little', nanosecond=False, bidirectional=True, analyse=False)[source]

Bases: TraceFlowBase[_DT, _BT, _IT, _PT], Generic[_DT, _BT, _IT, _PT]

Base flow tracing class.

Example

Registration is opt-in. Pass keyword argument protocol at class definition to register the flow tracing class under that protocol name:

class MyProtocol(TraceFlow, protocol='my_protocol'):
    ...

Omit it and the subclass is not registered, which is how a class that is not meant to be selectable by name declines:

class MyMixin(TraceFlow):  # not registered
    ...

Such a class can still be registered later, on demand:

Extractor.register_traceflow('my_mixin', MyMixin)
Parameters:

  • fout (str | None) – output path

  • format (str | None) – output format

  • byteorder (Literal['little', 'big']) – output file byte order

  • nanosecond (bool) – output nanosecond-resolution file flag

  • bidirectional (bool) – trace both halves of a conversation as one flow

  • analyse (bool) – reassemble each flow’s application layer

See also

For more information on customisation and extension, please refer to Customisation & Extensions.

property name: str

Protocol name of current class.

Note

This property is also available as a class variable. Its value can be set by __protocol_name__ class attribute.

property protocol: Type[pcapkit.protocols.protocol.ProtocolBase]

Protocol of current class.

Note

This property is also available as a class variable. Its value can be set by __protocol_type__ class attribute.

property registry: python:dict[str, ModuleDescriptor[TraceFlowBase] | typing.Type[TraceFlowBase]]

Mapping of protocol names to flow tracing classes.

Note

This property is only available as a class variable, since it is defined on TraceFlowMeta. It reads __traceflow__, the single table every flow tracing registration lands in, so it is not a per-class mapping. It is not __output__, which is the separate output-dumper table this class also owns.

property index: tuple[_IT, ...]

_IT table for traced flow.

classmethod register_dumper(format, dumper, ext)

Register a new dumper class.

Notes

The full qualified class name of the new dumper class should be as {dumper.module}.{dumper.name}.

The overwrite guard fires only when the incumbent dumper differs from the replacement, so re-registering the exact same object is a silent no-op rather than a warning about nothing displaced – the identity guard GitHub issue #718 gave the code-keyed registrars, extended here by GitHub issue #739. __output__ maps each format to a (dumper, ext) pair, so the identity check compares the incumbent dumper (index 0), not the pair – a re-registration that only changes ext is still identity-equal on the dumper and stays silent, since the dumper is what “the same object” means here, not the pair as a whole. __output__ is also a collections.defaultdict, unlike the other three registrars this issue touches; dict.get() does not invoke the default factory the way cls.__output__[format] would, so it stays non-inserting here as well.

Parameters:
classmethod register_callback(callback, *, index=None)

Register callback function.

Parameters:

  • callback (Callable[[TypeVar(_IT, bound= Info)], None]) – callback function, which will be called when reassembled datagram is obtained, with the list of reassembled datagrams as its only argument

  • index (int | None) – index of datagram to be called

classmethod make_fout(fout='./tmp', fmt='pcap')

Make root path for output.

Parameters:

  • fout (str) – root path for output

  • fmt (str) – output format

Return type:

tuple[Type[Dumper], str | None]

Returns:

Dumper of specified format and file extension of output file.

Warns:

  • FormatWarning – If fmt is not supported.

  • FileWarning – If fout exists and fmt is None.

Raises:

FileExists – If fout exists and fmt is NOT None.

abstractmethod dump(packet)

Dump frame to output files.

Parameters:

packet (TypeVar(_PT, bound= Info)) – a flow packet (trace.tcp.packet)

abstractmethod trace(packet, *, output=False)

Trace packets.

Overloads:

  • self, packet (_PT), output (Literal[True]) → Dumper

  • self, packet (_PT), output (Literal[False]) → str

Parameters:
Returns:

If output is True, returns the initiated Dumper object, which will dump data to the output file named after the flow label; otherwise, returns the flow label itself.

finish()

Finalise every flow still being traced.

Called by Extractor._cleanup once the capture has been read to its end, which is the point at which a flow that was never superseded can be said to be over.

The base implementation does nothing, so a tracer that has no such notion – or an existing third-party subclass that predates this method – keeps working unchanged. submit() must remain able to report a flow that was never finalised, since nothing guarantees this is called: a tracer driven directly rather than through an Extractor never sees an end of capture.

Implementations must be idempotent: Extractor._cleanup can run more than once for one extraction.

abstractmethod submit()

Submit traced TCP flows.

Return type:

tuple[TypeVar(_IT, bound= Info), ...]

Returns:

Traced TCP flow (trace.tcp.index).

__output__: DefaultDict[str, tuple[ModuleDescriptor[Dumper] | Type[Dumper], str | None]]

DefaultDict[str, tuple[ModuleDescriptor[Dumper] | ~typing.Type[Dumper], str | None]]: Format dumper mapping for writing output files. The values should be a tuple representing the module name and class name, or a dictdumper.dumper.Dumper subclass, and corresponding file extension.

__callback_fn__: list[Callable[[_IT], None]]

List of callback functions upon reassembled datagram.

_buffer: dict[_DT, _BT]

Buffer field (trace.tcp.buffer).

Type:

dict[_DT, _BT]

_stream: list[_IT]

Stream index (trace.tcp.index).

Type:

list[_IT]

_bidir

Bidirectional tracing flag. If set to True, both halves of a conversation share one buffer entry, one label and one output file; otherwise each direction is a flow of its own.

Type:

bool

_fproot

Output root path.

Type:

str

_foutio

Dumper class.

Type:

Type[Dumper]

_fdpext

Output file extension.

Type:

Optional[str]

_endian

Output file byte order.

Type:

Literal[‘little’, ‘big’]

_nnsecd

Output nanosecond-resolution file flag.

Type:

bool

_analyse

Application-layer analysis flag. If set to True, each flow reassembles the payload it carries so that its packet can be read; otherwise no payload is buffered and packet is None.

Type:

bool

__call__(packet)

Dump frame to output files.

Parameters:

packet (TypeVar(_PT, bound= Info)) – a flow packet (trace.tcp.packet)

classmethod __init_subclass__(protocol=None, *args, **kwargs)[source]

Initialise subclass.

This method is to be used for registering the flow tracing class to Extractor class.

Parameters:

  • protocol (str | None) – Protocol name to register the subclass under, lowercased. None (the default) skips registration entirely.

  • *args (Any) – Arbitrary positional arguments.

  • **kwargs (Any) – Arbitrary keyword arguments.

Raises:

UnsupportedCall – If any unrecognised class keyword is given.

Registration is opt-in: the subclass is registered if and only if protocol is given. This is what lets a subclass decline registration rather than having to inherit TraceFlowBase to avoid it, and it matches EnumSchema.__init_subclass__, which has guarded on its own code keyword all along.

Note

__protocol_name__ is not an opt-in. It supplies the name the class reports, which it does whether or not the class is registered; only the keyword decides registration.

See also

For more details, please refer to pcapkit.foundation.extraction.Extractor.register_traceflow().

__protocol_name__: str

Protocol name of current reassembly object.

__protocol_type__: Type[ProtocolBase]

Protocol of current reassembly object.

Internal Definitions

class pcapkit.foundation.traceflow.traceflow.TraceFlowBase(fout, format, byteorder='little', nanosecond=False, bidirectional=True, analyse=False)[source]

Bases: Generic[_DT, _BT, _IT, _PT]

Base flow tracing class.

Parameters:

  • fout (str | None) – output path

  • format (str | None) – output format

  • byteorder (Literal['little', 'big']) – output file byte order

  • nanosecond (bool) – output nanosecond-resolution file flag

  • bidirectional (bool) – trace both halves of a conversation as one flow

  • analyse (bool) – reassemble each flow’s application layer

Note

This class is for internal use only. For customisation, please use TraceFlow instead.

class pcapkit.foundation.traceflow.traceflow.TraceFlowMeta(name, bases, namespace, /, **kwargs)[source]

Bases: ABCMeta

Meta class to add dynamic support to TraceFlow.

This meta class is used to generate necessary attributes for the TraceFlow class. It can be useful to reduce unnecessary registry calls and simplify the customisation process.

Type Variables

pcapkit.foundation.traceflow.traceflow._DT: Any

Buffer ID data structure.

pcapkit.foundation.traceflow.traceflow._BT: pcapkit.corekit.infoclass.Info

Buffer data structure.

pcapkit.foundation.traceflow.traceflow._IT: pcapkit.corekit.infoclass.Info

Index data structure.

pcapkit.foundation.traceflow.traceflow._PT: pcapkit.corekit.infoclass.Info

Packet data structure.