Customisation & Extensions¶
New protocols, layers, fields and file formats can all be added to
pcapkit from outside the library.
Customising Protocols¶
Protocol is the user-facing abstract base
class of a protocol, which parses and constructs its packets. The built-in
protocol classes and layer base 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 |
||||
LINUX_SLL Family |
||||
VLAN Family |
||||
Internet Layer
( |
IP Family |
|||
IPv6 Extension Header |
||||
IPsec Family |
||||
ICMP Family |
||||
IGMP Family |
||||
Transport Layer
( |
||||
Application Layer
( |
FTP Family |
|||
HTTP Family |
||||
RARP Family |
||||
SSL Family |
||||
Miscellaneous Protocols |
PCAP Format |
|||
PCAP-NG Format |
||||
New Protocol¶
A protocol in an existing layer subclasses that layer’s base class:
Link,
Internet,
Transport or
Application.
Important
Subclass Protocol directly only for a new
protocol stack, such as the miscellaneous protocols of
pcapkit.protocols.misc.
Each protocol also needs a schema and a data model:
the schema, a
Schemasubclass, lays out the fields that parse and construct a packet, each a class attribute set to aFieldinstance;the data model, a
Datasubclass, holds a parsed packet, each attribute a bare type annotation.
Then register the class through pcapkit.foundation.registry.protocols,
into the table of the protocol that carries it. A table belongs to the protocol
that reads it, and is keyed by that protocol’s own code for its payload:
flowchart TD
FR["Frame.\_\_proto\_\_ for PCAP,<br/>PCAPNG.\_\_proto\_\_ for PCAP-NG<br/>keyed by LinkType<br/>register\_linktype for both,<br/>register\_pcap or register\_pcapng for one"]
LK["Link.\_\_proto\_\_<br/>keyed by EtherType<br/>register\_ethertype<br/>a VLAN tag reads on through it"]
LB["Loopback.\_\_proto\_\_<br/>keyed by BSD address family<br/>register\_loopback"]
IN["Internet.\_\_proto\_\_<br/>keyed by TransType<br/>register\_transtype"]
IPX["IPX.\_\_proto\_\_<br/>keyed by IPX Packet Type<br/>register\_ipx"]
subgraph TR ["Transport layer tables"]
direction TB
TU["TCP.\_\_proto\_\_ and UDP.\_\_proto\_\_<br/>keyed by port<br/>register\_tcp, register\_udp,<br/>register\_apptype for either or both"]
SC["SCTP.\_\_proto\_\_<br/>keyed by the DATA chunk's PPID<br/>register\_sctp"]
DC["DCCP.\_\_proto\_\_<br/>keyed by port<br/>register\_dccp"]
TU ~~~ SC ~~~ DC
end
APP["an application layer protocol"]
FR -->|"a link layer protocol, e.g. Ethernet"| LK
FR -->|"Loopback"| LB
FR -->|"IPv4 or IPv6, for a raw IP link type"| IN
LK -->|"an internet layer protocol, e.g. IPv4"| IN
LK -->|"IPX"| IPX
LK -->|"RARP"| APP
LB -->|"IPv4 or IPv6"| IN
IN -->|"TCP, UDP, SCTP or DCCP"| TR
IN -->|"IPX_in_IP"| IPX
IN -->|"OSPF"| APP
TR & IPX -->|"a registered protocol"| APP
A miss in any table gives that table’s fallback, normally
Raw, as
Next-Layer Dispatch shows.
Samples¶
A sketch of a new internet layer protocol, modelled on IPv4, whose full
implementation is pcapkit.protocols.internet.ipv4.protocol:
from typing import TYPE_CHECKING
from pcapkit.protocols.internet.internet import Internet # import base class
from pcapkit.protocols.data import Data # import data class
from pcapkit.protocols.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. Constructing a
protocol with a keyword no signature declares raises
UnsupportedCall, so a misspelling fails
instead of building a wrong field. The accepted set, read by
inspect.signature(), is the union of the keyword-taking parameters of
make, read, pack, unpack, __post_init__ and __init__
across the MRO: Protocol.__post_init__ hands one **kwargs
to both construction and parse, so a keyword only read declares still
travels through make.
A keyword no signature can express is listed in Protocol.__keywords__ instead:
class MyIPv4(Internet[IPv4Data, IPv4Schema],
schema=IPv4Schema, data=IPv4Data):
#: Read out of ``**kwargs`` by name, as ``IPv6_Ext.read`` does with
#: ``alias``; unioned down the MRO.
__keywords__ = frozenset({'my_extra_keyword'})
def read(self, length=None, **kwargs):
extra = kwargs.get('my_extra_keyword')
...
None skips the check, and is meant only for a dispatcher whose real
signature belongs to a class chosen at call time, such as
HTTP, whose
make() builds the
version it is given. Unlike a set, None is not inherited, so a
dispatcher’s subclasses stay checked. Prefer a declared parameter to either,
since it also documents the keyword.
Parsing is unaffected. Protocol.from_data checks its caller’s keywords
as strictly, but only warns with
UnknownFieldWarning for a key from your
_make_data, a disagreement between two of your own mappings rather than a
caller’s typo.
Warning
The check lives in Protocol.__init__, where every producer’s
keywords converge, so it covers SomeProtocol(...) and the pack it
leads to, but not make called on a bare instance, such as
object.__new__(cls).make(...), which drops an undeclared keyword
silently. This package’s tests and HTTP.make call make that
way. Covering it would mean interposing on every make in the tree, which
this package deliberately does not do.
Note
Instead of a register_* call, pass code= at class definition; a class
with neither stays unregistered:
class MyIPv4(Internet[IPv4Data, IPv4Schema],
schema=IPv4Schema, data=IPv4Data,
code=EtherType.Internet_Protocol_version_4):
...
The key’s type picks the registry – an
EtherType member means
Link’s, a
TransType member
Internet’s, and so on – so a
raw int, such as a TCP or UDP port, names its registry explicitly:
code={TCP: 21}. An iterable of either form registers into several at
once. ProtocolBase.__init_subclass__ gives the full
rule, and
register_protocol_code()
runs underneath it.
Extending Existing Protocol¶
Several protocols carry extension points, such as the options of IPv4. To handle a new code, register your methods against the protocol’s extension point.
Warning
Do not subclass the protocol class for this, which may cause unexpected
errors; use the registry functions in
pcapkit.foundation.registry.protocols.
The extension points, with the registry function that writes each:
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 extraction, reassembly, flow tracing and the
registries behind them, each extensible as the protocols are.
Extractor Engines¶
pcapkit.foundation.engines holds the engines that read packets out of a
capture file. The shipped engines derive from
EngineBase; build your own on
Engine, which carries the
engine= registration hook.
The 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 may support less than Extractor
offers. PyPCAP and pcap-ct dissect no protocol, so they support neither
reassembly nor flow tracing; PyPCAPFile has no IPv6 decoder, so it
reassembles IPv4 and TCP but not IPv6. Each engine warns, or raises, for a
capability it cannot provide rather than silently producing an empty result.
The exceptions pass without a warning: PyPCAPFile and DPKT leave TCP
behind some tags, tunnels and extension headers out of TCP reassembly and
flow tracing, DPKT also leaves IPv4 in IPv6, and IPv6 in IPv4, out of IP
reassembly, and the Scapy engine reads IPv6 only as far as Scapy
dissects it. See
Engine Support for the full table, which names each
case.
Note
PyPCAP and pcap-ct are two distributions of one libpcap(3)
interface, both installing a top-level pcap module, and are two engines:
engine='pypcap' and engine='pcap_ct'. Both exist because upstream
PyPCAP cannot be installed on Python 3.12 or newer and pcap-ct can.
Engine Support says which to pick.
Samples¶
A sketch of a new engine on scapy, modelled on
Scapy, whose full implementation is
pcapkit.foundation.engines.scapy:
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: ``engine`` registers the class under the key ``Extractor(engine=...)``
# looks up; without it, the class is not selectable by name.
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: Imported here, not at module level, so the engine's
# library stays an optional dependency.
from scapy import sendrecv # import API entry point
self._expkg = sendrecv # store API entry point
self._extmp = None # iterator over the packets, set by run()
super().__init__(extractor) # initialise base class
# NOTE: Open the input through ``self._expkg`` and store an iterator
# over its packets in ``self._extmp``.
def run(self) -> 'None':
ext = self._extractor # get extractor instance
# NOTE: Replace the verbose handler only for ``verbose=True``; a
# callable ``verbose=`` is the user's own, already in ``ext._vfunc``.
if ext._vfunc_default: # check if the default verbose output is requested
# NOTE: pcapkit.toolkit bundles per-engine helpers;
# ``packet2chain`` renders a Scapy packet as its protocol chain.
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: Parse the global header with the built-in engine and, unless
# output is disabled, dump it and set ``ext.format``.
ext.record_header()
self._extmp = iter(self._expkg.sniff(offline=ext._ifnm))
# NOTE: Return the next packet from ``self._extmp``, handling its
# verbose output, file output, reassembly and flow tracing.
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: Dump the packet unless output is disabled.
frnum = f'Frame {ext._frnum}'
if not ext._flag_q:
info = packet2dict(packet) # convert packet to dict-like object
# NOTE: In split mode (``ext._flag_f``) ``ext._ofile`` is a dumper
# class, built per frame; otherwise it is one dumper instance.
if ext._flag_f:
ofile = ext._ofile(f'{ext._ofnm}/{frnum}.{ext._fext}')
ofile(info, name=frnum)
else:
ext._ofile(info, name=frnum)
# NOTE: Each toolkit helper converts the packet to its reassembly's
# format, or gives None for a packet it does not apply to.
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: Flow tracing, likewise.
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: Store the frame when storing is on (``ext._flag_d``).
if ext._flag_d:
ext._frame.append(packet)
return packet # return the extracted packet
Output Dumpers¶
pcapkit.dumpkit builds on DictDumper, which writes
the extracted packets to the output file, and whose own documentation covers its
dumpers. A new output format is a dumper subclass.
The output dumpers and their formats:
Dumper Format |
Dumper Class |
Output Extension |
|---|---|---|
No Output |
N/A |
|
PCAP ( |
|
|
XML ( |
|
|
JSON ( |
|
|
Text ( |
|
Samples¶
A sketch of a PCAP dumper, modelled on PCAPIO,
whose full implementation is pcapkit.dumpkit.pcap:
from typing import TYPE_CHECKING
from pcapkit.dumpkit.common import Dumper # import base class
from pcapkit.protocols.misc.pcap.header.protocol import Header # import PCAP header class
from pcapkit.protocols.misc.pcap.frame.protocol import Frame # import PCAP frame class
if TYPE_CHECKING:
from typing import Optional
# NOTE: ``fmt`` registers the dumper under that output format, with ``ext``
# its file extension; without ``fmt``, the class is not registered.
class MyDumper(Dumper, fmt='pcap', ext='.pcap'):
# NOTE: The format name, not the extension, which is ``ext`` above.
@property
def kind(self) -> 'str':
return 'pcap'
# NOTE: Take what the file header needs, e.g. the link-layer protocol,
# and keep any state later frames need.
def __init__(self, fname: 'str', *, ...) -> 'None':
self._fnum = 1 # initialise frame number counter to 1
super().__init__(fname, ...) # initialise base class
# NOTE: Overridden because PCAP is binary, so the file reopens in binary
# rather than 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: Called on initialisation, to write the file header.
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: Write one packet, here as a PCAP frame.
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
self._fnum += 1 # increment frame number counter
Reassembly and Flow Tracing¶
pcapkit.foundation.reassembly reassembles fragmented packets;
pcapkit.foundation.traceflow follows the packets of a flow. The shipped
classes derive from
ReassemblyBase and
TraceFlowBase; build your own on
Reassembly or
TraceFlow, which carry the
protocol= registration hook.
The reassembly and flow tracing classes, per protocol:
Protocol |
Reassembly Class |
Flow Tracing Class |
|---|---|---|
See also
pcapkit.foundation.reassembly and pcapkit.foundation.traceflow
document the classes and their algorithms.
New Implementation¶
A sketch of an IPv4 reassembly class, modelled on
IPv4, whose full implementation is
pcapkit.foundation.reassembly.ipv4; a flow tracing class subclasses
TraceFlow the same way:
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.protocol 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: First the data models the class is generic over.
BufferID: 'TypeAlias' = ... # key of a reassembly table entry, e.g.
# (src, dst, id, proto) for IPv4 and IPv6,
# (src, srcport, dst, dstport) for TCP
@info_final
class Packet(Info):
# NOTE: An extracted packet, as the reassembly reads it.
...
@info_final
class DatagramID(Info):
# NOTE: The key identifying a reassembled datagram.
...
@info_final
class Datagram(Info):
# NOTE: A reassembled datagram.
...
@info_final
class Buffer(Info):
# NOTE: A reassembly table entry, buffering fragments until complete.
...
# NOTE: ``protocol`` registers the class under that key; without it, the
# class is not selectable by name.
class MyReassembly(Reassembly[Packet, Datagram, BufferID, Buffer],
protocol='ipv4'):
__protocol_name__ = 'IPv4' # name of the protocol
__protocol_type__ = IPv4 # type of the protocol
# NOTE: Buffer ``info`` in ``self._buffer``, a dict of ``BufferID`` to
# ``Buffer``.
def reassembly(self, info: 'Packet') -> 'None':
... # reassembly algorithm implementation
# NOTE: Once a buffer is complete, store its datagrams.
self._dtgram.extend(
self.submit(...) # submit reassembled datagram
)
# NOTE: Convert a complete ``Buffer`` into ``Datagram`` instances.
def submit(self, buf: 'Buffer', *, bufid: 'BufferID') -> 'list[Datagram]':
datagram = ... # list of ``Datagram`` instances
# NOTE: Run the registered callbacks on the submission.
for callback in self.__callback_fn__:
callback(datagram) # run callback functions
return datagram
Callback Functions¶
A callback registered on a reassembly or flow tracing class inspects the datagrams or 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. Each should return
None; any return value is ignored.
A reassembly callback can discard datagrams by modifying its list in place, but prefer not to: if the result itself is wrong for your purpose, subclass the class and change its algorithm instead.