Customisation & Extensions¶
pcapkit is built for customisation: new protocols, layers, fields and
even new file formats can all be added from outside the library.
What’s in for Protocols?¶
Protocol is the most essential concept
and class in pcapkit: the user-facing abstract base class for a
protocol, which is responsible for parsing and/or constructing the protocol
packets, as in the network stack. The built-in protocol classes derive from
its internal base, ProtocolBase,
rather than from Protocol itself.
The protocol classes pcapkit ships:
Protocol Type |
Protocol Class |
|||
|---|---|---|---|---|
Link Layer
( |
ARP Family |
|||
L2TP Family |
||||
VLAN Family |
||||
Internet Layer
( |
IP Family |
|||
IPv6 Extension Header |
||||
IPsec Family |
||||
Transport Layer
( |
||||
Application Layer
( |
FTP Family |
|||
HTTP Family |
||||
RARP Family |
||||
Miscellaneous Protocols |
PCAP Format |
|||
PCAP-NG Format |
||||
New Protocol¶
To add a new protocol to an existing layer, you need to create a new class
inherited from that layer’s base class, specifically, the
Link,
Internet,
Transport, and
Application classes, which
are the base classes for link, internet, transport, and application layer
protocols, respectively.
Important
The Protocol class is the user-facing
abstract base class for protocols, and it is not recommended to inherit from
it directly, unless you are going to create a new protocol stack, e.g., a new
miscellaneous protocol stack as in pcapkit.protocols.misc module. The
four layer base classes above, like the built-in miscellaneous protocols,
derive from ProtocolBase rather than
from Protocol.
Besides, you need to implement the corresponding schema and data model for the
new protocol. The schema is a Schema
subclass that defines the field structure of the protocol, and the data model
is a Data subclass that defines the data
attributes of the protocol.
Note
The schema classes are used for parsing and/or constructing the protocol
packets, each field should be defined as a class attribute with the value of
a Field subclass instance.
The data classes are used for storing the parsed protocol packets. Each data attribute should be defined as a class attribute with no value besides the type annotations of the corresponding field.
Once the class exists, register it through
pcapkit.foundation.registry.protocols, picking the registry that matches
the protocol’s layer – a link layer protocol goes to
register_linktype(), and so on.
The registry function per protocol type:
Protocol Type |
Registry Function |
Notes |
|
|---|---|---|---|
Link Layer |
Registry functions on the left column are combined calls to those on the right columns. It is recommended to use those on the left columns when registering new protocols. |
||
Internet Layer |
|||
Transport Layer |
|||
Application Layer |
|||
Samples¶
The following code snippet shows how to create a new protocol class:
Note
Following is a PoC implementation of a new internet layer protocol. For
demonstration purpose, we use the IPv4 protocol as an example, where the
comprehensive implementation can be found in the
pcapkit.protocols.internet.ipv4 module.
from typing import TYPE_CHECKING
from pcapkit.protocols.internet.internet import Internet # import base class
from pcapkit.protocols.data.data import Data # import data class
from pcapkit.protocols.schema.schema import Schema # import schema class
from pcapkit.corekit.fields.ipaddress import IPv4AddressField # import field class
from pcapkit.const.reg.ethertype import EtherType # import protocol code registry
from pcapkit.foundation.registry.protocols import register_ethertype # import protocol registry API
if TYPE_CHECKING:
from ipaddress import IPv4Address
from typing import Any, Optional
class IPv4Data(Data): # define data model
...
src: 'IPv4Address' # source address
dst: 'IPv4Address' # destination address
...
class IPv4Schema(Schema): # define fields
...
src = IPv4AddressField(...)
dst = IPv4AddressField(...)
...
class MyIPv4(Internet[IPv4Data, IPv4Schema],
schema=IPv4Schema, data=IPv4Data): # define protocol class
def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'IPv4Data':
if length is None:
length = len(self) # infer length of packet from data
schema = self.__header__ # get schema
return IPv4Data(...)
def make(self, ..., **kwargs: 'Any') -> 'IPv4Schema':
...
return IPv4Schema(...)
# register protocol class
register_ethertype(EtherType.Internet_Protocol_version_4, MyIPv4)
Important
Declare every construction keyword your protocol accepts. Building a
protocol through its constructor with a keyword that no signature
declares raises UnsupportedCall rather than
discarding it silently, so a misspelling costs an exception instead of a
silently wrong field. The accepted set is read from inspect.signature() – the union of every
keyword-taking parameter of make, read, pack, unpack,
__post_init__ and __init__ anywhere in the class’s MRO – which is
wider than make alone because ProtocolBase.__post_init__ hands the same
**kwargs to the construction and to the parse, so a keyword only read
declares still travels through make.
Two shapes a signature cannot express are declared on the class instead, via
ProtocolBase.__keywords__:
class MyIPv4(Internet[IPv4Data, IPv4Schema],
schema=IPv4Schema, data=IPv4Data):
#: A keyword read out of ``**kwargs`` by name rather than declared as a
#: parameter, as ``ESP.read`` does with ``packet``. Unioned down the MRO.
__keywords__ = frozenset({'my_extra_keyword'})
def read(self, length=None, **kwargs):
extra = kwargs.get('my_extra_keyword')
...
Setting it to None skips the check entirely, and is meant only for a
dispatcher whose real signature belongs to a class chosen at call time –
HTTP.make is the one
such class in the library. Unlike a set, None is not inherited, so a
subclass of a dispatcher is checked normally. Prefer a declared parameter to
either: it is also what documents the keyword to your callers.
Parsing is unaffected. ProtocolBase.from_data checks the keywords its
caller passes just as strictly, raising
UnsupportedCall, but only warns
UnknownFieldWarning for a key that comes
from your _make_data – a mismatch there is a disagreement between two of
your own mappings rather than a caller’s typo.
Warning
The check lives in ProtocolBase.__init__, where every producer’s
keywords converge, so it covers SomeProtocol(...) and the pack it
leads to – but not a direct SomeProtocol.make(...) call, which still
discards an undeclared keyword in silence. object.__new__(cls).make(...)
is the idiom that reaches it, used by this package’s own tests and by
HTTP.make to reach its
versioned implementation. Covering that would mean interposing on every
make in the tree – a larger change this package deliberately does not
make. Construct through the constructor to get the check.
Note
Registering after the fact, as above, is one option. The other is passing
code= at class definition, which is opt-in – a class that omits it
stays unregistered, exactly as if no register_* call had been made
either:
class MyIPv4(Internet[IPv4Data, IPv4Schema],
schema=IPv4Schema, data=IPv4Data,
code=EtherType.Internet_Protocol_version_4):
...
The key’s own type decides which registry it goes to – an
EtherType member always means
Link, a
TransType member always means
Internet, and so on – so a
raw int, such as a TCP or UDP port, cannot say which registry it
belongs to and must be given explicitly: code={TCP: 21}. Either form
may appear in an iterable to register one class into several registries
at once. See
ProtocolBase.__init_subclass__ for the full
rule, and
register_protocol_code() for
what runs underneath it.
Extending Existing Protocol¶
Several protocols carry customisable attributes – the option headers of IPv4, for one. To add handling for one of them, check the protocol class for the extension point you need, then register your methods against it.
Warning
Do not inherit from the protocol class directly – it may cause unexpected
errors. Use the helper registry functions in
pcapkit.foundation.registry.protocols instead.
The extension points of each protocol class, with the registry function that writes them:
Protocol Type |
Protocol Class |
Extensible Registry |
Registry Function |
|---|---|---|---|
Internet Layer |
|||
Transport Layer |
|||
Application Layer |
|||
Miscellaneous Protocols |
|||
Important
Every extension point takes a class method: its first argument is the protocol class, and the rest match the built-in methods it sits beside.
And Speaking of Library Foundations…¶
pcapkit.foundation carries the library’s own machinery – extraction,
reassembly, flow tracing and the registries behind them – and each part of it
is extensible in the same way the protocols are.
Extractor Engines¶
pcapkit.foundation.engines holds the built-in engines that read packets
out of an input PCAP, PCAP-NG or other capture file. The shipped engines derive
from EngineBase;
Engine is the subclass to build your
own on, since it is the one carrying the engine= registration hook.
The available engines and the file formats they read:
Engine Type |
Engine Class |
Supported File Formats |
|---|---|---|
Built-in Engines |
PCAP only |
|
PCAP-NG only |
||
Third-party Engines |
all formats supported by Scapy |
|
all formats supported by DPKT |
||
all formats supported by PyShark |
||
PCAP only, and from a file on disk |
||
PCAP only, and from a file on disk |
||
PCAP only |
Note
An engine is free to support less than Extractor
offers, and several do. PyPCAP and pcap-ct perform no protocol dissection
whatsoever, so they support neither reassembly nor flow tracing; PyPCAPFile
has no IPv6 decoder, so it supports IPv4 and TCP reassembly but not IPv6. What
matters is that the gap is announced – each engine warns, or raises, for the
capability it cannot provide, rather than silently producing an empty result,
save that PyPCAPFile leaves TCP behind a VLAN tag, in a tunnel or behind an
extension header (an IPv6 one, or AH in IPv4) out without a warning; that
DPKT leaves TCP in an IP-in-IP tunnel or behind an AH or IPv6 extension
header in IPv4, and IPv4 in IPv6 or IPv6 in IPv4, out without one; and the
Scapy engine reads IPv6 only as far as Scapy dissects it, which leaves
IPv6 behind some BSD loopback headers, and TCP and IPv6 fragments behind some
IPv6 extension headers, out without one. See
Engine Support for the full table.
Note
PyPCAP and pcap-ct are two distributions of one libpcap(3)
interface, and both install a top-level pcap module, so they are two
engines rather than one: engine='pypcap' and engine='pcap_ct'. Upstream
PyPCAP cannot be installed on Python 3.12 or newer, and pcap-ct can –
which is why both exist. See
Engine Support for which to pick.
Samples¶
The following code snippet shows how to create a new engine class:
Note
Following is a PoC implementation of a new engine based on scapy. For documentation
purposes, we use the Scapy engine as an example,
where the comprehensive implementation can be found in the pcapkit.foundation.engines.scapy
module.
from typing import TYPE_CHECKING
from pcapkit.foundation.engines.engine import Engine # import base class
if TYPE_CHECKING:
from pcapkit.foundation.extraction import Extractor
from scapy.packet import Packet
# NOTE: The ``engine`` keyword is what registers the engine with the Extractor,
# and it is the key ``Extractor(engine=...)`` will look it up under. It is
# required: registration is opt-in, so omitting it defines a perfectly usable
# class that is simply not selectable by name. Note that __engine_name__ is
# *not* an opt-in -- it sets the name the engine reports about itself, which
# it does whether or not the engine is registered.
class MyScapy(Engine['Packet'], engine='scapy'):
__engine_name__ = 'Scapy' # friendly name of the engine
__engine_module__ = 'scapy' # module name that the engine is based on
def __init__(self, extractor: 'Extractor') -> 'None':
# NOTE: the API entry point of the underlying engine module should
# be imported here, instead of at the top of the module, to avoid
# dependency issues.
from scapy import sendrecv # import API entry point
self._expkg = sendrecv # store API entry point
self._extmp = None # intermediate storage for the iterator
# generated by the API entry point
super().__init__(extractor) # initialise base class
# NOTE: The following methods are the core methods of the engine class,
# which are responsible for parsing the input file and extracting the
# network packets for further processing. It is expected to create an
# iterator instance that yields the network packets, and store it in
# the intermediate storage, i.e., self._extmp for further processing,
# using the API entry point as in the self._expkg attribute.
def run(self) -> 'None':
ext = self._extractor # get extractor instance
# NOTE: You may wish to do some pre-processing here, e.g., to
# check if there's any extraction protocol and/or layer thresholds
# set by the user (some engines may not support this feature); or
# to set up the verbose handler for the underlying engine module,
# as in the following code snippet.
if ext._flag_v: # check if verbose mode is enabled
# NOTE: PyPCAPKit has its own bundled supporting functions for
# third-party libraries, e.g., the `packet2chain` function is
# to convert a given Packet instance of the Scapy library to a
# human-readable string, which is used for verbose output.
from pcapkit.toolkit.scapy import packet2chain # import utility function
ext._vfunc = lambda extractor, packet: print(
f'Frame {extractor._frnum:>3d}: {packet2chain(packet)}'
) # print verbose message as `Frame XXX: Ethernet:IP:TCP:...`
# NOTE: The extractor's shared global-header step parses the global
# header with the built-in engine and, unless output is disabled,
# dumps it and sets the output format that `ext.format` reports.
ext.record_header()
# NOTE: Here we use the API entry point, i.e., scapy.sendrecv.sniff,
# to read the input file and extract the network packets. The API
# entry point returns an iterable instance, which is then converted
# as an iterator and stored in the intermediate storage for further
# processing.
self._extmp = iter(self._expkg.sniff(offline=ext._ifnm))
# NOTE: The following methods are the helper methods of the engine class,
# which are responsible for extracting the network packets from the
# intermediate storage, i.e., self._extmp, and returning the extracted
# network packets for further processing. It is also expected to handle
# necessary actions, including but not limited to, verbose output, file
# output, reassembly and flow tracing, etc.
def read_frame(self) -> 'Packet':
from pcapkit.toolkit.scapy import (ipv4_reassembly, ipv6_reassembly, packet2dict,
tcp_reassembly, tcp_traceflow) # import utility functions
ext = self._extractor # get extractor instance
packet = next(self._extmp) # get next packet from the iterator
ext._frnum += 1 # increment frame number
ext._vfunc(ext, packet) # print verbose message
# NOTE: The following code snippet is to handle the file output,
# i.e., to write the packet to the output file if the output file
# is specified by the user.
frnum = f'Frame {ext._frnum}'
if not ext._flag_q:
info = packet2dict(packet) # convert packet to dict-like object
# NOTE: Now, we need to check if output file is to be a single
# file. If so, we can directly write the packet to the output
# file, i.e., ext._ofile; otherwise, we need to create a new
# dumper instance by initialising ext._ofile, and write the
# packet to the dumper instance. For more information, please
# refer to the documentation of DictDumper library.
if ext._flag_f:
ofile = ext._ofile(f'{ext._ofnm}/{frnum}.{ext._fext}')
ofile(info, name=frnum)
else:
ext._ofile(info, name=frnum)
# NOTE: The following code snippet is to handle the reassembly of
# the network packets, i.e., to reassemble the fragmented packets
# into a complete packet. Before reassembly, we need to convert
# the packet to a tailored format, i.e., a dict-like object, which
# is then passed to the reassembly function. For more information,
# please refer to the documentation of pcapkit.foundation.reassembly
# module.
if ext._flag_r:
if ext._ipv4:
data_ipv4 = ipv4_reassembly(packet, count=ext._frnum)
if data_ipv4 is not None:
ext._reasm.ipv4(data_ipv4)
if ext._ipv6:
data_ipv6 = ipv6_reassembly(packet, count=ext._frnum)
if data_ipv6 is not None:
ext._reasm.ipv6(data_ipv6)
if ext._tcp:
data_tcp = tcp_reassembly(packet, count=ext._frnum)
if data_tcp is not None:
ext._reasm.tcp(data_tcp)
# NOTE: The following code snippet is to handle the flow tracing of
# the network packets, i.e., to trace a series of packets as they
# will compose as a flow and/or stream. Before reassembly, we need
# to convert the packet to a tailored format, i.e., a dict-like object,
# which is then passed to the flow tracing function. For more
# information, please refer to the documentation of
# pcapkit.foundation.traceflow module.
if ext._flag_t:
if ext._tcp:
data_tf_tcp = tcp_traceflow(packet, count=ext._frnum)
if data_tf_tcp is not None:
ext._trace.tcp(data_tf_tcp)
# NOTE: The following code snippet is to record extracted frames
# into the internal storage, i.e., ext._frame, for further processing.
if ext._flag_d:
ext._frame.append(packet)
return packet # return the extracted packet
Output Dumpers¶
pcapkit.dumpkit wraps DictDumper, which writes the
extracted packets to the output file. New output formats are added by
subclassing a dumper, on the same extensibility DictDumper
itself offers.
See also
DictDumper’s own documentation covers the dumpers in
detail.
The available output dumpers and their formats:
Dumper Format |
Dumper Class |
Output Extension |
|---|---|---|
No Output |
N/A |
|
PCAP ( |
|
|
XML ( |
|
|
JSON ( |
|
|
Text ( |
|
Samples¶
The following code snippet shows how to create a new dumper class:
Note
Following is a PoC implementation of a new dumper for the PCAP format. For
documentation purposes, we use the PCAPIO
dumper as an example, where the comprehensive implementation can be found
in the pcapkit.dumpkit.pcap module.
from typing import TYPE_CHECKING
from pcapkit.dumpkit.common import Dumper # import base class
from pcapkit.protocols.misc.pcap.header import Header # import PCAP header class
from pcapkit.protocols.misc.pcap.frame import Frame # import PCAP frame class
if TYPE_CHECKING:
from typing import Optional
# NOTE: We need to specify the output file format and extension here at the
# class definition, as in the following code snippet. Such that PyPCAPKit
# will be able to automatically register our dumper class to the corresponding
# dumper registries, and use it to dump the extracted network packets.
class MyDumper(Dumper, fmt='pcap', ext='.pcap'):
# NOTE: This property names the file format of the dumper, as a string.
# It is not the file extension -- that is the ``ext`` given at class
# definition above.
@property
def kind(self) -> 'str':
return 'pcap'
# NOTE: We need to collect some necessary information here thru the necessary
# arguments, e.g., the file name, link-layer protocol, etc., which will be
# used to generate the top-level header of the output file.
def __init__(self, fname: 'str', *, ...) -> 'None':
# NOTE: We can do some pre-processing here if necessary, e.g., to
# save the nanosecond-resolution flag, link-layer protocol type, etc.,
# such that we can use them later when dumping the network packets.
self._fnum = 1 # initialise frame number counter to 1
super().__init__(fname, ...) # initialise base class with the file name
# and other necessary arguments
# NOTE: We need to overwrite the original ``__call__`` method, cause the
# PCAP file format is a binary format, thus we need to re-open the output
# file in binary mode, instead of the default text mode.
def __call__(self, value: 'Frame', name: 'Optional[str]' = None) -> 'MyDumper':
with open(self._file, 'ab') as file:
self._append_value(value, file, name or f'Frame {self._fnum}') # append value to the file
return self
# NOTE: The following method will be called at the initialisation of the
# dumper class, which is expected to write the top-level header of the
# output file.
def _dump_header(self, *, ...) -> 'None':
packet = Header(...) # create PCAP header instance based on the given arguments
with open(self._file, 'wb') as file:
file.write(packet.data) # write the header to the output file
# NOTE: The following method is to write the network packets to the output
# file. It is expected to convert the network packet to a tailored format,
# i.e., a Frame instance, which is then converted to the binary format and
# written to the output file.
def _append_value(self, value: 'Frame', file: 'IO[bytes]', name: 'str') -> 'None':
packet = Frame(...) # create PCAP frame instance based on the given arguments
file.write(packet.data) # write the frame to the output file
# NOTE: We can do some post-processing here if necessary, e.g., to
# update the frame number counter, etc., such that we can use them
# later when dumping other network packets.
self._fnum += 1 # increment frame number counter
Reassembly and Flow Tracing¶
pcapkit.foundation.reassembly reassembles fragmented packets;
pcapkit.foundation.traceflow follows packets that compose a flow or
stream. The shipped classes derive from
ReassemblyBase and
TraceFlowBase;
Reassembly and
TraceFlow are the ones to build
your own on, since they carry the protocol= registration hook.
The reassembly and flow tracing classes, per protocol:
Protocol |
Reassembly Class |
Flow Tracing Class |
|---|---|---|
See also
Please refer to the documentation of pcapkit.foundation.reassembly and
pcapkit.foundation.traceflow modules for more information about the
reassembly and flow tracing classes, as well as the underlying algorithms.
New Implementation¶
A new reassembly or flow tracing class subclasses
Reassembly or
TraceFlow:
Note
Following is a PoC implementation of a new reassembly class for the IPv4
protocol. For documentation purposes, we use the
IPv4 reassembly class as an
example, where the comprehensive implementation can be found in the
pcapkit.foundation.reassembly.ipv4 module.
from typing import TYPE_CHECKING
from pcapkit.corekit.infoclass import Info, info_final # import Info class for data model definitions
from pcapkit.protocols.internet.ipv4 import IPv4 # import protocol class for attribute definitions
from pcapkit.foundation.reassembly.reassembly import Reassembly # import base class
if TYPE_CHECKING:
from typing import TypeAlias
# NOTE: We need to specify some useful data structure before creating the
# reassembly class, e.g., the data model for the reassembled packet, the
# data model for the reassembly table, etc., which will be used to store
# the reassembled packet and the reassembly table, respectively.
BufferID: 'TypeAlias' = ... # a tuple-like object to identify the reassembly
# table entry's ID, e.g. (src, dst, id, proto)
# for both IPv4 and IPv6, and
# (src, srcport, dst, dstport) for TCP
@info_final
class Packet(Info):
# NOTE: This is the data model for the extracted packet, which is
# expected to be a dict-like object, and will be used to store the
# information from the extracted packets to be reassembled.
...
@info_final
class DatagramID(Info):
# NOTE: This is the data model for the reassembled datagram's ID,
# which is expected to be a dict-like object, and will be used to
# store the information to identify the reassembled datagram.
...
@info_final
class Datagram(Info):
# NOTE: This is the data model for the reassembled datagram, which
# is expected to be a dict-like object, and will be used to store
# the information of the reassembled datagram.
...
@info_final
class Buffer(Info):
# NOTE: This is the data model for the reassembly table, which is
# expected to be a dict-like object, and will be used to store the
# information of each buffered entry in the reassembly table, that
# is to be reassembled with the following fragmented packets.
...
# NOTE: Now we can create the reassembly class, which is expected to be
# a subclass of the base class, i.e., Reassembly, and implement the core
# methods, i.e., reassembly and submit, for reassembling the fragmented
# packets and submitting the reassembled datagram, respectively.
#
# The ``protocol`` keyword is what registers the class, and it is the key it
# will be looked up under. It is required: registration is opt-in, so omitting
# it defines a perfectly usable class that is simply not selectable by name.
# Note that __protocol_name__ is *not* an opt-in -- it sets the name the class
# reports about itself, registered or not. Note also that the keyword is
# spelled ``protocol`` here and ``engine`` on Engine above.
class MyReassembly(Reassembly[Packet, Datagram, BufferID, Buffer],
protocol='ipv4'):
__protocol_name__ = 'IPv4' # name of the protocol
__protocol_type__ = IPv4 # type of the protocol
# NOTE: This method is to reassemble the fragmented packets, i.e.,
# ``info`` formatted as a ``Packet`` instance, into the reassembly
# table, i.e., ``self._buffer`` formatted as a dict of ``BufferID``
# mapped to ``Buffer`` instances.
def reassembly(self, info: 'Packet') -> 'None':
... # reassembly algorithm implementation
# NOTE: Once the fragmented packets are reassembled, we need to
# submit the reassembled datagram, i.e., ``Datagram`` instance,
# to the storage, i.e., ``self._dtgram``, for further processing.
self._dtgram.extend(
self.submit(...) # submit reassembled datagram
)
# NOTE: This method is to convert the reassembled datagram, i.e.,
# ``buf`` formatted as a ``Buffer`` instance, to a ``Datagram``
# instance, which is then submitted to the storage, i.e.,
# ``self._dtgram``.
def submit(self, buf: 'Buffer', * , bufid: 'BufferID') -> 'list[Datagram]':
datagram = ... # submit algorithm implementation, e.g., to convert
# the reassembled datagram to a list of ``Datagram``
# instances, which is then returned for further
# processing by the caller
# NOTE: Before we submit the reassembled datagram, we need to
# run the registered callback functions, if any, to process the
# reassembled datagram, e.g., to check if the reassembled datagram
# is to be discarded, etc.
for callback in self.__callback_fn__:
callback(datagram) # run callback functions
return datagram
Callback Functions¶
Callback functions can be registered on the reassembly and flow tracing classes, to inspect the reassembled datagrams or traced flows as they are produced.
See also
pcapkit.foundation.registry.foundation holds the callback registry
functions:
A reassembly callback runs on every submission and takes one argument, the list
of datagrams that submission produced. A flow tracing callback runs as each flow
is finalised and takes that flow’s
Index. Either returns
None; any return value is ignored.
A reassembly callback can modify its list in place, which discards datagrams, but prefer not to: if the reassembly or tracing result itself is wrong for your purpose, subclass the reassembly or flow tracing class and change its algorithm instead.