Follow TCP Stream#

pcapkit.foundation.traceflow.tcp is the interface to trace TCP flows from a series of packets and connections.

class pcapkit.foundation.traceflow.tcp.TCP(fout, format, byteorder='little', nanosecond=False)[source]#

Bases: TraceFlowBase[tuple[_AT, int, _AT, int], Buffer, Index, Packet[_AT]], Generic[_AT]

Trace TCP flows.

Parameters:

fout (Optional[str]) – output path
format (Optional[str]) – output format
byteorder (Literal['little', 'big']) – output file byte order
nanosecond (bool) – output nanosecond-resolution file flag
*args – Arbitrary positional arguments.
**kwargs – Arbitrary keyword arguments.

dump(packet)[source]#

Dump frame to output files.

Parameters:: packet (Packet[TypeVar(_AT, IPv4Address, IPv6Address)]) – a flow packet (trace.tcp.packet)
Return type:: None

trace(packet, *, output=False)[source]#

Trace packets.

Parameters:

packet (Packet[TypeVar(_AT, IPv4Address, IPv6Address)]) – a flow packet (trace.tcp.packet)
output (bool) – flag if has formatted dumper

Return type:

Dumper | str

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.

Notes

The flow label is formatted as following:

f'{packet.src}_{packet.srcport}-{packet.dst}_{info.dstport}-{packet.timestamp}'

submit()[source]#

Submit traced TCP flows.

Return type:: tuple[Index, ...]
Returns:: Traced TCP flow (trace.tcp.index).

__protocol_name__: str = 'TCP'#: Protocol name of current reassembly object.

__protocol_type__(file=None, length=None, **kwargs): Type[Protocol] = <class 'pcapkit.protocols.transport.tcp.TCP'>#: Protocol of current reassembly object.

Terminology#

trace.tcp.packet#

Data structure for TCP flow tracing (TraceFlow.dump) is as following:

tract_dict = dict(
    protocol=data_link,                     # data link type from global header
    index=frame.info.number,                # frame number
    frame=frame.info,                       # extracted frame info
    syn=tcp.flags.syn,                      # TCP synchronise (SYN) flag
    fin=tcp.flags.fin,                      # TCP finish (FIN) flag
    src=ip.src,                             # source IP
    dst=ip.dst,                             # destination IP
    srcport=tcp.srcport,                    # TCP source port
    dstport=tcp.dstport,                    # TCP destination port
    timestamp=frame.info.time_epoch,        # frame timestamp
)

trace.tcp.buffer#

Data structure for internal buffering when performing flow tracing algorithms (TraceFlow._buffer) is as following:

(dict) buffer --> memory buffer for reassembly
 |--> (tuple) BUFID : (dict)
 |       |--> ip.src      |
 |       |--> tcp.srcport |
 |       |--> ip.dst      |
 |       |--> tcp.dstport |
 |                        |--> 'fpout' : (dictdumper.dumper.Dumper) output dumper object
 |                        |--> 'index': (list) list of frame index
 |                        |              |--> (int) frame index
 |                        |--> 'label': (str) flow label generated from ``BUFID``
 |--> (tuple) BUFID ...

trace.tcp.index#

Data structure for TCP flow tracing (element from TraceFlow.index tuple) is as following:

(tuple) index
 |--> (Info) data
 |     |--> 'fpout' : (Optional[str]) output filename if exists
 |     |--> 'index': (tuple) tuple of frame index
 |     |              |--> (int) frame index
 |     |--> 'label': (str) flow label generated from ``BUFID``
 |--> (Info) data ...

Data Structures#

class pcapkit.foundation.traceflow.data.tcp.Packet(*args: VT, **kwargs: VT)[source]#

Bases: Info, Generic[_AT]

Data structure for TCP flow tracing.

See also

pcapkit.foundation.traceflow.TraceFlow.dump()
trace.tcp.packet

protocol: Enum_LinkType#: Data link type from global header.

index: int#: Frame number.

frame: Data_Frame | dict[str, Any]#: Extracted frame info.

syn: bool#: TCP synchronise (SYN) flag.

fin: bool#: TCP finish (FIN) flag.

src: _AT#: Source IP.

dst: _AT#: Destination IP.

srcport: int#: TCP source port.

dstport: int#: TCP destination port.

timestamp: float#: Frame timestamp.

pcapkit.foundation.traceflow.data.tcp.BufferID#

Buffer ID.

alias of tuple[_AT, int, _AT, int]

class pcapkit.foundation.traceflow.data.tcp.Buffer(*args: VT, **kwargs: VT)[source]#

Bases: Info

Data structure for TCP flow tracing.

See also

pcapkit.foundation.traceflow.TraceFlow.index
trace.tcp.buffer

fpout: Dumper#: Output dumper object.

index: list[int]#: List of frame index.

label: str#: Flow label generated from BUFID.

class pcapkit.foundation.traceflow.data.tcp.Index(*args: VT, **kwargs: VT)[source]#

Bases: Info

Data structure for TCP flow tracing.

See also

element from pcapkit.foundation.traceflow.TraceFlow.index tuple
trace.tcp.index

fpout: Optional[str]#: Output filename if exists.

index: tuple[int, ...]#: Tuple of frame index.

label: str#: Flow label generated from BUFID.

Type Variables#

pcapkit.foundation.traceflow.data.tcp._AT: ipaddress.IPv4Address | ipaddress.IPv6Address#