Root Protocol

pcapkit.protocols.protocol contains Protocol only, which is an abstract base class for all protocol family, with pre-defined utility arguments and methods of specified protocols.

class pcapkit.protocols.protocol.Protocol(file=None, length=None, **kwargs)[source]

Bases: ProtocolBase, Generic[_PT, _ST]

Abstract base class for all protocol family.

abstract property name: str

Name of current protocol.

property alias: str

Acronym of current protocol.

property info_name: str

Key name of the info dict.

property info: _PT

Info dict of current instance.

property data: bytes

Binary packet data of current instance.

abstract property length: int

Header length of current protocol.

property payload: ProtocolBase

Payload of current instance.

property protocol: str | None

Name of next layer protocol (if any).

property protochain: ProtoChain

Protocol chain of current instance.

property packet: Packet

Data_Packet data of the protocol.

Note

The split relies on self.length being the length of the octets preceding the payload, and on the payload running from there to the end of the buffer. Both hold for a protocol laid out as a header followed by its payload, which is nearly all of them.

A protocol that is not laid out that way has to override this: one whose length counts something else, or one carrying a trailer after the payload, gets a header that eats the payload and a payload of b''. That is what PCAPNG did to every packet block – its length is the wire’s Block Total Length and the captured octets sit ahead of the option list and the trailing length field – and ProtocolBase.__init__() injects this payload into every parsed _info, so the empty value reached the dumpers and corrupted the files they wrote. See #646.

property schema: _ST

Schema data of the protocol.

classmethod id()

Index ID of the protocol.

Return type:

tuple[str, ...]

Returns:

By default, it returns the name of the protocol. In certain cases, the method may return multiple values.

See also

pcapkit.protocols.protocol.Protocol.__getitem__()

classmethod register(code, protocol)

Register a new protocol class.

Notes

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

Parameters:
Raises:

pcapkit.utilities.exceptions.RegistryError – If protocol is not a ProtocolBase subclass.

Warns:

pcapkit.utilities.warnings.RegistryWarning – If code is already registered. The warning names the displaced entry and its replacement, so a caller can tell what was lost rather than only that something was.

Note

The guard now matches register_protocol’s: it fires only when the incumbent differs from the replacement, so re-registering the exact same class object under the same code is a silent no-op rather than a warning about nothing displaced. GitHub issue #718 corrected the previous presence-only guard here, which read every repeat registration as a caller mistake even when the value was unchanged. The identity check does not reintroduce the concern that guard was written to avoid: it is a plain is comparison, so an incumbent left as an unresolved ModuleDescriptor is never equal to the resolved replacement without the descriptor being resolved – the comparison itself resolves nothing, so the deferred import stays deferred and such an incumbent still reports as different.

classmethod analyze(proto, payload, **kwargs)

Analyse packet payload.

Parameters:

  • proto (int) – Protocol registry number.

  • payload (bytes) – Packet payload.

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

Return type:

ProtocolBase

Returns:

Parsed payload as a Protocol instance.

classmethod from_schema(schema)

Create protocol instance from schema.

Parameters:

schema (TypeVar(_ST, bound= Schema) | dict[str, Any]) – Protocol schema.

Return type:

Self

Returns:

Protocol instance.

classmethod from_data(data)

Create protocol instance from data.

Parameters:

data (TypeVar(_PT, bound= Data) | dict[str, Any]) – Protocol data.

Return type:

Self

Returns:

Protocol instance.

abstractmethod read(length=None, **kwargs)

Read (parse) packet data.

Parameters:

  • length (int | None) – Length of packet data.

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

Return type:

TypeVar(_PT, bound= Data)

Returns:

Parsed packet data.

abstractmethod make(**kwargs)

Make (construct) packet data.

Parameters:

**kwargs (Any) – Arbitrary keyword arguments.

Return type:

TypeVar(_ST, bound= Schema)

Returns:

Curated protocol schema data.

Note

The **kwargs here absorbs the keywords that ProtocolBase.__post_init__ hands to the parse as well as to the construction, so an implementation is not expected to declare every keyword it is called with. It is not a place for a caller to put a keyword no signature declares: since #617, building a protocol through its constructor with such a keyword raises UnsupportedCall from ProtocolBase.__init__ rather than discarding it.

Warning

Calling this method directly is not checked, and still discards an undeclared keyword in silence. The check lives in ProtocolBase.__init__, so it covers SomeProtocol(...) and the pack() it leads to, but not SomeProtocol.make(...) on an instance obtained some other way – object.__new__(cls).make(**kwargs) is the idiom, used by this package’s own tests and by HTTP.make to reach its versioned implementation. Covering it would mean interposing on every make in the tree rather than on the one place their keywords converge, which is a larger change than #617 and deliberately not made here. Construct through the constructor to get the check.

unpack(length=None, **kwargs)

Unpack (parse) packet data.

Parameters:

  • length (int | None) – Length of packet data.

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

Return type:

TypeVar(_PT, bound= Data)

Returns:

Parsed packet data.

Notes

We used a special keyword argument __packet__ to pass the global packet data to underlying methods. This is useful when the packet data is not available in the current instance.

pack(**kwargs)

Pack (construct) packet data.

Parameters:

**kwargs (Any) – Arbitrary keyword arguments.

Return type:

bytes

Returns:

Constructed packet data.

Notes

We used a special keyword argument __packet__ to pass the global packet data to underlying methods. This is useful when the packet data is not available in the current instance.

static decode(byte, *, encoding=None, errors='strict')

Decode bytes into str.

Should decoding failed using encoding, the method will try again decoding the bytes as 'unicode_escape' with 'replace' for error handling.

See also

The method is a wrapping function for bytes.decode().

Parameters:

  • byte (bytes) – Source bytestring.

  • encoding (str | None) – The encoding with which to decode the bytes. If not provided, pcapkit will first try detecting its encoding using chardet. The fallback encoding would is UTF-8.

  • errors (Literal['strict', 'ignore', 'replace']) – The error handling scheme to use for the handling of decoding errors. The default is 'strict' meaning that decoding errors raise a UnicodeDecodeError. Other possible values are 'ignore' and 'replace' as well as any other name registered with codecs.register_error() that can handle UnicodeDecodeError.

Return type:

str

static unquote(url, *, encoding='utf-8', errors='replace')

Unquote URLs into readable format.

Should decoding failed , the method will try again replacing '%' with '\x' then decoding the url as 'unicode_escape' with 'replace' for error handling.

See also

This method is a wrapper function for urllib.parse.unquote().

Parameters:

  • url (str) – URL string.

  • encoding (str) – The encoding with which to decode the bytes.

  • errors (Literal['strict', 'ignore', 'replace']) – The error handling scheme to use for the handling of decoding errors. The default is 'strict' meaning that decoding errors raise a UnicodeDecodeError. Other possible values are 'ignore' and 'replace' as well as any other name registered with codecs.register_error() that can handle UnicodeDecodeError.

Return type:

str

static expand_comp(value)

Expand protocol class to protocol name.

The method is used to expand protocol class to protocol name, in the following manner:

  1. If value is a protocol instance, the method will return the protocol class, and the protocol names in upper case obtained from Protocol.id.

  2. If value is a protocol class, the method will return the protocol class itself, and the protocols names in upper case obtained from Protocol.id.

  3. If value is str, the method will attempt to search for the existing registered protocol class from pcapkit.protocols.__proto__ and follow step 2; otherwise, return the value itself.

Parameters:

value (str | ProtocolBase | Type[ProtocolBase]) – Protocol class or name.

Return type:

tuple

__layer__: Literal['Link', 'Internet', 'Transport', 'Application'] | None

Layer of protocol, can be one of Link, Internet, Transport and Application. For example, the layer of Ethernet is Link. However, certain protocols are not in any layer, such as Raw, and thus its layer is None.

__proto__: DefaultDict[int, ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]

Protocol index mapping for decoding next layer, c.f. self._decode_next_layer & self._import_next_layer. The values should be a tuple representing the module name and class name, or a Protocol subclass.

__schema__: Type[_ST]

Protocol header schema definition.

__header__: _ST

Protocol header schema instance.

_read_packet(length=None, *, header=None, payload=None, discard=False)

Read raw packet data.

Overloads:

  • self, length (Optional[int]), header (None) → bytes

  • self, header (int), payload (Optional[int]), discard (Literal[True]) → bytes

  • self, header (int), payload (Optional[int]), discard (Literal[False]) → Data_Packet

Parameters:

  • length (int | None) – length of the packet

  • header (int | None) – length of the packet header

  • payload (int | None) – length of the packet payload

  • discard (bool) – flag if discard header data

  • If header omits, returns the whole packet data in bytes.

  • If discard is set as True, returns the packet body (in bytes) only.

  • Otherwise, returns the header and payload data as Packet object.

_get_payload()

Get payload from self.__header__.

Return type:

bytes

Returns:

Payload of self.__header__ as bytes.

See also

This is a wrapper function for pcapkit.protocols.schema.schema.Schema.get_payload().

classmethod _make_data(data)

Create key-value pairs from data for protocol construction.

Parameters:

data (Data) – protocol data

Return type:

dict[str, Any]

Returns:

Key-value pairs for protocol construction.

classmethod _make_index(name, default=None, *, namespace=None, reversed=False, pack=False, size=4, signed=False, lilendian=False)

Return first index of name from a dict or enumeration.

Overloads:

  • cls, name (int | StdlibEnum | AenumEnum), pack (Literal[False]) → int

  • cls, name (int | StdlibEnum | AenumEnum), pack (Literal[True]), size (int), signed (bool), lilendian (bool) → bytes

  • cls, name (str), default (Optional[int]), namespace (Type[StdlibEnum] | Type[AenumEnum]), pack (Literal[False]) → int

  • cls, name (str), default (Optional[int]), namespace (Type[StdlibEnum] | Type[AenumEnum]), pack (Literal[True]), size (int), signed (bool), lilendian (bool) → bytes

  • cls, name (str), default (Optional[int]), namespace (dict[int, str]), reversed (Literal[False]), pack (Literal[False]) → int

  • cls, name (str), default (Optional[int]), namespace (dict[int, str]), reversed (Literal[False]), pack (Literal[True]), size (int), signed (bool), lilendian (bool) → bytes

  • cls, name (str), default (Optional[int]), namespace (dict[str, int]), reversed (Literal[True]), pack (Literal[False]) → int

  • cls, name (str), default (Optional[int]), namespace (dict[str, int]), reversed (Literal[True]), pack (Literal[True]), size (int), signed (bool), lilendian (bool) → bytes

  • cls, name (str | int | StdlibEnum | AenumEnum), default (Optional[int]), namespace (Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]), reversed (bool), pack (Literal[False]) → int

Parameters:
Returns:

Index of name from a dict or enumeration. If pack is True, returns bytes; otherwise, returns int.

Raises:

ProtocolNotImplemented – If name is NOT in namespace and default is None.

classmethod _make_payload(data)

Create payload from data for protocol construction.

This method uses __next_type__ and __next_name__ to determine the payload type and name. If either of them is None, a NoPayload instance will be returned. Otherwise, the payload will be constructed by Protocol.from_data.

Parameters:

data (Data) – protocol data

Return type:

ProtocolBase

Returns:

Payload for protocol construction.

static _lookup_registry(registry, code)

Look up a dispatch registry entry without recording a miss.

Parameters:

  • registry (DefaultDict[Any, TypeVar(_VT)]) – dispatch registry to read, i.e. self.__proto__ or one of the per-protocol __option__ / __chunk__ / __block__ family. Passed in rather than read from the class, so that a caller reaching the registry through an instance keeps doing so.

  • code (Any) – registry key to look up, i.e. the wire code being dispatched on

Return type:

TypeVar(_VT)

Returns:

The entry registered for code, or the fallback registry declares when code is not registered.

Important

Every one of these registries is a collections.defaultdict held on a class attribute, shared by every instance of the class in the process. So registry[code] inserts each code it misses, and parsing one packet carrying an unrecognised code is enough to grow the registry permanently.

The inserted value is whatever the default factory would have produced anyway, so the entry buys nothing. It costs a spurious “already registered” warning from the next genuine register call for that code, and it makes “is this code registered?” unanswerable by inspection, since the answer depends on what has been parsed. The fallback is therefore read from the default factory directly rather than through a lookup that records it.

static _lookup_next_layer(registry, proto)

Look up the protocol class registered for a next layer code.

Parameters:
Return type:

Type[ProtocolBase]

Returns:

The class registered for proto, or the fallback registry declares – normally Raw – when proto is not registered.

Important

The lookup itself is self._lookup_registry, so a miss does not grow the shared registry. What this adds is the next-layer-specific resolution step: a registered code may hold a ModuleDescriptor rather than a class, and importing it is written back so the import happens once.

That write-back is deliberately confined to a hit. Memoising the fallback’s resolution under proto would be exactly the insertion self._lookup_registry exists to avoid.

So a miss resolves its fallback descriptor again on every frame, and what keeps that affordable is ModuleDescriptor.klass reading sys.modules instead of re-entering importlib.import_module() – see #574. Memoising the resolved class here instead, whether under proto, in registry’s default factory, or in a cache beside the registry, would retain a class that importlib.reload() then makes stale; #425 and #428 at this layer and #560 at the schema layer are all that same defect.

_decode_next_layer(dict_, proto, length=None, *, packet=None)

Decode next layer protocol.

Parameters:

  • dict_ (TypeVar(_PT, bound= Data)) – info buffer

  • proto (int) – next layer protocol index

  • length (int | None) – valid (non-padding) length

  • packet (dict[str, Any] | None) – packet info (passed from self.unpack)

Return type:

TypeVar(_PT, bound= Data)

Returns:

Current protocol with next layer extracted.

Notes

We added a new key __next_type__ to dict_ to store the next layer protocol type, and a new key __next_name__ to store the next layer protocol name. These two keys will NOT be included when Info.to_dict is called.

_import_next_layer(proto, length=None, *, packet=None)

Import next layer extractor.

Parameters:
Return type:

ProtocolBase

Returns:

Instance of next layer.

_get_context(cls=None)

Get the caller supplied context for this protocol, if any.

The lookup is keyed on self.id, so a protocol finds its own context without knowing how the caller spelled the registry.

Parameters:

cls (Type[TypeVar(_CTX, bound= ProtocolContext)] | None) – Expected context class; when given, a context registered under this protocol’s name but of another type is ignored rather than returned for the implementation to trip over.

Return type:

TypeVar(_CTX, bound= ProtocolContext) | None

Returns:

The matching context, or None when the caller supplied none.

See also

pcapkit.corekit.context

_data: bytes

Raw packet data.

_file: IO[bytes]

Source packet stream.

_info: _PT

Parsed packet data.

_next: ProtocolBase

Next layer protocol instance.

_protos: ProtoChain

Protocol chain instance.

_seekset: int

File pointer.

Type:

int

_sigterm

If terminate parsing next layer of protocol.

Type:

bool

__data__: Type[_PT] = <class 'pcapkit.protocols.data.misc.raw.Raw'>

Protocol packet data definition.

__init__(file=None, length=None, **kwargs)

Initialisation.

Overloads:

  • self, file (IO[bytes] | bytes), length (Optional[int]), kwargs (Any) → None

  • self, kwargs (Any) → None

Parameters:

  • file (IO[bytes] | bytes | None) – Source packet stream.

  • length (int | None) – Length of packet data.

  • _layer (str) – Parse packet until _layer (self._exlayer). While parsing, the un-prefixed layer is accepted as well – see the note below.

  • _protocol (Union[str, Protocol, Type[Protocol]]) – Parse packet until _protocol (self._exproto). While parsing, the un-prefixed protocol is accepted as well – see the note below.

  • packet (dict[str, Any]) – Packet context of the enclosing layer, as handed over by self._import_next_layer. While parsing, it is republished as __packet__ so that self.unpack – and through it the schema – can see it; see the note below.

  • __context__ (Union[ContextRegistry, ProtocolContext, Mapping[str, ProtocolContext], Iterable[ProtocolContext]]) – Caller supplied parsing context (self._exctx), c.f. pcapkit.corekit.context. It is consumed here rather than being forwarded to self.read, and is propagated to nested layers by self._import_next_layer.

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

Raises:

UnsupportedCall – When constructing (file is None), if a keyword names no parameter of this protocol’s make(), read(), pack(), unpack(), __post_init__() or __init__(), anywhere in the MRO, and is not listed in __keywords__. See #617; until then such a keyword was silently discarded. Parsing (file is given) is unaffected.

Note

Three of the keywords above are out-of-band: they configure the parse rather than describing the packet, and every one of them is consumed here, at the one point each of a protocol’s producers passes through. That is deliberate, and it is what the normalisation below relies on – fixing a spelling here fixes it for the engines, for all four _import_next_layer implementations, and for any third party protocol that copied their shape, rather than one call site at a time.

__post_init__(file=None, length=None, **kwargs)

Post initialisation hook.

Overloads:

  • self, file (IO[bytes] | bytes), length (Optional[int]), kwargs (Any) → None

  • self, kwargs (Any) → None

Parameters:

  • file (IO[bytes] | bytes | None) – Source packet stream.

  • length (int | None) – Length of packet data.

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

See also

For construction arguments, please refer to self.make.

classmethod __init_subclass__(schema=None, data=None, code=None, *args, **kwargs)[source]

Initialisation for subclasses.

Parameters:

  • schema (Type[TypeVar(_ST, bound= Schema)] | None) – Schema class.

  • data (Type[TypeVar(_PT, bound= Data)] | None) – Data class.

  • code (Any) – Next-layer dispatch registration key(s). None (the default) skips registration entirely. See ProtocolBase.__init_subclass__() for the accepted shapes and the enum-type inference rule.

  • *args (Any) – Arbitrary positional arguments.

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

This method is called when a subclass of Protocol is defined. It is used to set the self.__schema__ attribute of the subclass.

Notes

When schema and/or data is not specified, the method will first try to find the corresponding class in the schema and data modules respectively. If the class is not found, the default Raw and Raw classes will be used.

This method also registers the subclass to the protocol registry, i.e., pcapkit.protocols.__proto__. That registration is unconditional – it is the name-keyed identity registry, unrelated to the code keyword – whereas code’s next-layer dispatch registration is opt-in; see ProtocolBase.__init_subclass__() for the latter.

See also

For more information on the registry, please refer to pcapkit.foundation.registry.protocols.register_protocol().

__repr__()

Returns representation of parsed protocol data.

Example

>>> protocol
<Frame alias='...' frame=(..., packet=b'...', sethernet=..., protocols='Ethernet:IPv6:Raw')>
Return type:

str

__str__()

Returns formatted hex representation of source data stream.

Example

>>> protocol
Frame(..., packet=b"...", sethernet=..., protocols='Ethernet:IPv6:Raw')
>>> print(protocol)
00 00 00 00 00 00 00 a6 87 f9 27 93 16 ee fe 80 00 00 00     ..........'........
00 00 00 1c cd 7c 77 ba c7 46 b7 87 00 0e aa 00 00 00 00     .....|w..F.........
fe 80 00 00 00 00 00 00 1c cd 7c 77 ba c7 46 b7 01 01 a4     ..........|w..F....
5e 60 d9 6b 97                                               ^`.k.
Return type:

str

__getitem__(key)

Subscription (getitem) support.

  • If key is a Protocol object, the method will fetch its indexes (self.id).

  • Later, search the packet’s chain of protocols with the calculated key.

  • If no matches, then raises ProtocolNotFound.

Parameters:

key (str | Protocol | Type[Protocol]) – Indexing key.

Return type:

ProtocolBase

Returns:

The sub-packet from the current packet of indexed protocol.

Raises:

ProtocolNotFound – If key is not in the current packet.

See also

The method calls self.expand_comp to handle the key and expand it for robust searching.

__contains__(name)

Returns if certain protocol is in the instance.

Parameters:

name (str | Protocol | Type[Protocol]) – Name to search

See also

The method calls self.expand_comp to handle the name and expand it for robust searching.

Return type:

bool

abstractmethod classmethod __index__()

Numeral registry index of the protocol.

Return type:

IntEnum | IntEnum

_exlayer: str | None

Parse packet until such layer.

Type:

str

_exproto: str | ProtocolBase | Type[ProtocolBase] | None

Parse packet until such protocol.

Type:

str

_exctx: ContextRegistry | None = None

Caller supplied parsing context, c.f. pcapkit.corekit.context. self.__init__ replaces this with a real ContextRegistry; the class level None is what an instance built without going through __init__ – e.g. object.__new__(SomeProtocol) – sees, so that reading it is always safe.

Data Models

class pcapkit.protocols.data.protocol.Packet(*args: VT, **kwargs: VT)[source]

Bases: Data

Header and payload data.

header: bytes

packet header

payload: bytes

packet payload

Internal Definitions

class pcapkit.protocols.protocol.ProtocolBase(file=None, length=None, **kwargs)[source]

Bases: Generic[_PT, _ST]

Abstract base class for all protocol family.

Note

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

class pcapkit.protocols.protocol.ProtocolMeta(name, bases, namespace, /, **kwargs)[source]

Bases: ABCMeta

Meta class to add dynamic support to Protocol.

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

Type Variables

pcapkit.protocols.protocol._PT: pcapkit.protocols.data.data.Data
pcapkit.protocols.protocol._ST: pcapkit.protocols.schema.schema.Schema