Protocol Family¶

pcapkit.protocols is collection of all protocol families, with detailed implementation and methods.

All protocol classes are implemented as ProtocolBase subclasses, which are responsible for processing extracted binary packet data and/or construct protocol packet from given information. Below is a brief diagram of the class hierarchy of pcapkit.protocols:

        flowchart LR
    A{{ProtocolMeta}} -.->|metaclass| B(ProtocolBase)

    subgraph link [Link Layer]
        Link --> Ethernet & L2TP & OSPF & VLAN & ARP

        subgraph arp [ARP Family]
            ARP --> InARP & RARP

            subgraph rarp [RARP Family]
                RARP --> DRARP
            end
        end

        subgraph vlan [VLAN Family]
            VLAN --> C_Tag & S_Tag
        end

        subgraph l2tp [L2TP Family]
            L2TP --> L2TPv2
        end
    end

    subgraph internet [Internet Layer]
        Internet --> HIP & IPX & IP & ipv6ext

        subgraph ip [IP Family]
            IP --> IPv4 & IPv6 & IPsec

            subgraph ipsec [IPsec Family]
                IPsec --> AH & ESP
            end
        end

        subgraph ipv6ext [IPv6 Extension Header]
            IPv6-Frag & IPv6-Opts & IPv6-Route & HOPOPT & MH
        end
    end

    subgraph transport [Transport Layer]
        Transport --> TCP & UDP & SCTP
    end

    subgraph application [Application Layer]
        Application --> HTTP & FTP

        subgraph http [HTTP Family]
            HTTP --> h1["HTTP/1.*"] & h2["HTTP/2"]
        end

        subgraph ftp [FTP Family]
            FTP & FTP_DATA
        end
    end

    subgraph misc [Miscellaneous]
        subgraph pcap [PCAP Format]
            Header & Frame
        end

        subgraph pcapng [PCAP-NG Format]
            PCAPNG
        end

        Raw & NoPayload
    end

    B --> Link & Internet & Transport & Application

    B --> Header & Frame & PCAPNG & Raw & NoPayload
    Raw --> FTP_DATA

    B --> C(Protocol)
    C --> D([user customisation ...])

    click A "/pcapkit/protocols/protocol.html#pcapkit.protocols.protocol.ProtocolMeta"
    click B "/pcapkit/protocols/protocol.html#pcapkit.protocols.protocol.ProtocolBase"
    click C "/pcapkit/protocols/protocol.html#pcapkit.protocols.protocol.Protocol"
    click D "/ext.html#what-s-in-for-protocols"

    click Link "/pcapkit/protocols/link/link.html#pcapkit.protocols.link.Link"
    click Ethernet "/pcapkit/protocols/link/ethernet.html#pcapkit.protocols.link.ethernet.Ethernet"
    click L2TP "/pcapkit/protocols/link/l2tp.html#pcapkit.protocols.link.l2tp.L2TP"
    click L2TPv2 "/pcapkit/protocols/link/l2tpv2.html#pcapkit.protocols.link.l2tpv2.L2TPv2"
    click OSPF "/pcapkit/protocols/link/ospf.html#pcapkit.protocols.link.ospf.OSPF"
    click VLAN "/pcapkit/protocols/link/vlan.html#pcapkit.protocols.link.vlan.VLAN"
    click C_Tag "/pcapkit/protocols/link/c_tag.html#pcapkit.protocols.link.c_tag.C_Tag"
    click S_Tag "/pcapkit/protocols/link/s_tag.html#pcapkit.protocols.link.s_tag.S_Tag"
    click ARP "/pcapkit/protocols/link/arp.html#pcapkit.protocols.link.arp.ARP"
    click InARP "/pcapkit/protocols/link/arp.html#pcapkit.protocols.link.arp.InARP"
    click RARP "/pcapkit/protocols/link/rarp.html#pcapkit.protocols.link.rarp.RARP"
    click DRARP "/pcapkit/protocols/link/rarp.html#pcapkit.protocols.link.rarp.DRARP"

    click Internet "/pcapkit/protocols/internet/internet.html#pcapkit.protocols.internet.Internet"
    click AH "/pcapkit/protocols/internet/ah.html#pcapkit.protocols.internet.ah.AH"
    click ESP "/pcapkit/protocols/internet/esp.html#pcapkit.protocols.internet.esp.ESP"
    click HIP "/pcapkit/protocols/internet/hip.html#pcapkit.protocols.internet.hip.HIP"
    click HOPOPT "/pcapkit/protocols/internet/hopopt.html#pcapkit.protocols.internet.hopopt.HOPOPT"
    click IP "/pcapkit/protocols/internet/ip.html#pcapkit.protocols.internet.ip.IP"
    click IPsec "/pcapkit/protocols/internet/ipsec.html#pcapkit.protocols.internet.ipsec.IPsec"
    click IPv4 "/pcapkit/protocols/internet/ipv4.html#pcapkit.protocols.internet.ip.ipv4.IPv4"
    click IPv6 "/pcapkit/protocols/internet/ipv6.html#pcapkit.protocols.internet.ip.ipv6.IPv6"
    click IPv6-Frag "/pcapkit/protocols/internet/ipv6_frag.html#pcapkit.protocols.internet.ipv6_frag.IPv6_Frag"
    click IPv6-Opts "/pcapkit/protocols/internet/ipv6_opts.html#pcapkit.protocols.internet.ipv6_opts.IPv6_Opts"
    click IPv6-Route "/pcapkit/protocols/internet/ipv6_route.html#pcapkit.protocols.internet.ipv6_route.IPv6_Route"
    click IPX "/pcapkit/protocols/internet/ipx.html#pcapkit.protocols.internet.ipx.IPX"
    click MH "/pcapkit/protocols/internet/mh.html#pcapkit.protocols.internet.mh.MH"

    click Transport "/pcapkit/protocols/transport/transport.html#pcapkit.protocols.transport.Transport"
    click TCP "/pcapkit/protocols/transport/tcp.html#pcapkit.protocols.internet.tcp.TCP"
    click UDP "/pcapkit/protocols/transport/udp.html#pcapkit.protocols.internet.udp.UDP"
    click SCTP "/pcapkit/protocols/transport/sctp.html#pcapkit.protocols.transport.sctp.SCTP"

    click Application "/pcapkit/protocols/application/application.html#pcapkit.protocols.application.Application"
    click HTTP "/pcapkit/protocols/application/http.html#pcapkit.protocols.application.http.HTTP"
    click h1 "/pcapkit/protocols/application/httpv1.html#pcapkit.protocols.application.httpv1.HTTP"
    click h2 "/pcapkit/protocols/application/httpv2.html#pcapkit.protocols.application.httpv2.HTTP"
    click FTP "/pcapkit/protocols/application/ftp.html#pcapkit.protocols.application.ftp.FTP"
    click FTP_DATA "/pcapkit/protocols/application/ftp.html#pcapkit.protocols.application.ftp.FTP_DATA"

    click Raw "/pcapkit/protocols/misc/raw.html#pcapkit.protocols.misc.raw.Raw"
    click NoPayload "/pcapkit/protocols/misc/null.html#pcapkit.protocols.misc.null.NoPayload"
    click PCAPNG "/pcapkit/protocols/misc/pcapng.html#pcapkit.protocols.misc.pcapng.PCAPNG"
    click Header "/pcapkit/protocols/misc/pcap.html#pcapkit.protocols.misc.pcap.header.Header"
    click Frame "/pcapkit/protocols/misc/pcap.html#pcapkit.protocols.misc.pcap.frame.Frame"

Protocol Registry¶

pcapkit.protocols.__proto__: dict[str, Type[ProtocolBase]]¶

Protocol registry.

Type:: dict[str, Type[Protocol]]

Header Schema¶

class pcapkit.protocols.schema.schema.Schema(dict_=None, **kwargs)[source]¶

Bases: Mapping[str, _VT], Generic[_VT]

Schema for protocol headers.

__payload__: str = 'payload'¶: Field name of the payload.

__additional__: list[str]¶: List of additional built-in names.

__excluded__: list[str]¶: List of names to be excluded from dict conversion.

classmethod __init_subclass__(*args, **kwargs)[source]¶

Refuse to derive from a finalised schema.

final() is a promise to the type checker and nothing more: it records __final__ on the class and leaves the interpreter free to subclass it anyway. Every schema schema_final() finalises carries a generated __init__ built from the __fields__ that were declared at that moment, so a subclass adding a field afterwards inherits a constructor that cannot set it – which is #422’s failure shape, reached by a different route. This turns that promise into a rule the interpreter keeps.

Parameters:

*args (Any) – Arbitrary positional arguments.
**kwargs (Any) – Arbitrary keyword arguments in class definition.

Raises:

SchemaError – If any class in cls’s ancestry carries the __final__ marker in its own __dict__.

static __new__(cls, *args, **kwargs)[source]¶

Create a new instance.

The class will try to automatically generate __init__ method with the same signature as specified in class variables’ type annotations, which is inspired by PEP 557 (dataclasses).

Parameters:

*args (TypeVar(_VT)) – Arbitrary positional arguments.
**kwargs (TypeVar(_VT)) – Arbitrary keyword arguments.

Raises:

SchemaError – If cls was marked with final() but never finalised by schema_final(), i.e. it carries __final__ in its own __dict__ without FINAL.

Return type:: Self Out of scope, deliberately – see pcapkit.corekit.infoclass.Info.__new__(), which documents the identical gap: a @final class descending from a BASE ancestor inherits BASE rather than NONE and so never reaches the branch below at all, regardless of its own marker. No BASE-state schema in the tree is marked @final by hand today, so the escape is theoretical rather than live.

pack(packet=None)[source]¶

Pack Schema into bytes.

Parameters:: packet (dict[str, Any] | None) – Packet data.
Return type:: bytes
Returns:: Packed Schema as bytes.

Notes

Since we do not know the length of the packet, we use a reasonable default value -1 for the __length__ field, as the Field class will consider negative value as a placeholder.

If you want to pack the packet with the correct length, please provide the __length__ value before packing.

pre_pack(packet)[source]¶

Prepare packet data for packing process.

Parameters:: packet (dict[str, Any]) – packet data

Note

This method is expected to directly modify any data stored in the packet and thus no return is required.

classmethod unpack(cls, data, length=None, packet=None)[source]¶

Unpack bytes into Schema.

Parameters:

data (bytes | IO[bytes]) – Packed data.
length (int | None) – Length of data.
packet (dict[str, Any] | None) – Unpacked data.

Return type:

Self

Returns:

Unpacked data as Schema.

Notes

We used a __length__ key in packet to record the length of the remaining data, which is used to determine the length of the payload field.

When this schema is nested – unpacked through a SchemaField rather than directly – packet is not the enclosing schema’s own data, but a context built by nested_packet_context(): a name this schema does not itself declare falls through to the enclosing schema, and the enclosing schema is also reachable unconditionally under a __packet__ key. See that function for the exact lookup, write and iteration semantics.

And an __option_padding__ key in the packet to record how much of an OptionField’s declared area it did not consume. What follows that area decides what the remainder means: usually padding, to be skipped, but a schema may equally read it as further options, as the PCAP-NG name resolution block does with its records and options.

An OptionField declares the size of the whole area it may read, but stops at the end-of-option-list marker and reports the unconsumed remainder through __option_padding__. Since that remainder has not been parsed, we rewind data by it, so that the fields which size themselves from __option_padding__ read the remainder itself rather than the same number of octets from beyond it.

classmethod pre_unpack(packet)[source]¶

Prepare packet data for unpacking process.

Parameters:: packet (dict[str, Any]) – packet data

Note

This method is expected to directly modify any data stored in the packet and thus no return is required.

post_process(packet)[source]¶

Revise schema data after packing and/or unpacking process.

Parameters:: packet (dict[str, Any]) – Unpacked data.
Return type:: Schema
Returns:: Revised schema.

get_payload(name=None)[source]¶

Get payload of Schema.

Parameters:: name (str | None) – Name of the payload field.
Return type:: bytes
Returns:: Payload of Schema as bytes.

classmethod from_dict(dict_=None, **kwargs)[source]¶

Create a new instance.

If dict_ is present and has a .keys() method, then does: for k in dict_: self[k] = dict_[k].
If dict_ is present and has no .keys() method, then does: for k, v in dict_: self[k] = v.
If dict_ is not present, then does: for k, v in kwargs.items(): self[k] = v.

Parameters:

dict_ (Mapping[str, TypeVar(_VT)] | Iterable[tuple[str, TypeVar(_VT)]] | None) – Source data.
**kwargs (TypeVar(_VT)) – Arbitrary keyword arguments.

Return type:

Self

to_dict()[source]¶

Convert Schema into dict.

Important

We only convert nested Schema objects into dict if they are the direct value of the Schema object’s attribute. Should such Schema objects be nested within other data, types, such as list, tuple, set, etc., we shall not convert them into dict and remain them intact.

Return type:: dict[str, TypeVar(_VT)]

to_bytes()[source]¶

Convert Schema into bytes.

Return type:: bytes

__fields__: OrderedDict[str, FieldBase]¶: Mapping of fields.

class pcapkit.protocols.schema.schema.EnumSchema(dict_=None, **kwargs)[source]¶

Bases: Schema, Generic[_ET]

Schema with enumeration mapping support.

Examples

To create an enumeration mapping supported schema, simply

class MySchema(EnumSchema[MyEnum]):

    # optional, set the default schema for enumeration mapping
    # if the enumeration number is not found in the mapping
    __default__ = lambda: UnknownSchema  # by default, None

then, you can use inheritance to create a list of schemas for this given enumeration mapping:

class OneSchema(MySchema, code=MyEnum.ONE):
    ...

class MultipleSchema(MySchema, code=[MyEnum.TWO, MyEnum.THREE]):
    ...

or optionally, using the register() method to register a schema to the enumeration mapping:

MySchema.register(MyEnum.ZERO, ZeroSchema)

And now you can access the enumeration mapping via the registry property (more specifically, class attribute):

>>> MySchema.registry[MyEnum.ONE]  # OneSchema

__default__: Callable[[], Type[Self]]¶: Callback to return the default schema for enumeration mapping, by default is a lambda: None statement.

__enum__: DefaultDict[_ET, Type[Self]]¶: Mapping of enumeration numbers to schemas.

property registry: DefaultDict[_ET, Type[Self]]¶: Mapping of enumeration numbers to schemas.

Note

This property is also available as a class attribute.

Important

See EnumMeta.registry: the returned mapping is a _EnumRegistry, so looking up an unregistered code returns the default schema without recording code as if it had been registered.

classmethod register(code, schema)[source]¶

Parameters:

code (TypeVar(_ET, bound= Enum)) – Enumetaion code.
schema (Type[Self]) – Enumetaion schema.

Warns:

pcapkit.utilities.warnings.RegistryWarning – If code is already registered, naming the displaced schema and its replacement.

Note

Every public registrar in pcapkit.foundation.registry.protocols that accepts a schema registers two halves of one binding – a parser class through e.g. IPv4.register_option, and a schema class through this method. The parser half has warned on an overwrite for as long as it has existed; this half assigned bare, so one register_ipv4_option call replacing a built-in reported the parser it displaced and said nothing about the schema. The guard here closes that asymmetry.

It fires only when the incumbent differs from the replacement, the same guard register_protocol applies, even though code here – unlike register_protocol’s key – is supplied by the caller and independent of schema. GitHub issue #718 corrected the previous presence-only guard: a repeat call that names the exact same schema object is a caller replaying a registration, not a mistake, so it is now a silent no-op.

Presence is a faithful “was this really registered” test only because _EnumRegistry returns a miss without recording it. A plain collections.defaultdict would have inserted __default__ the first time any unregistered code was looked up, so parsing a single packet carrying an unknown code would have made the next legitimate registration for that code warn about an entry no caller ever asked for – the defect fixed for this layer in #555, and for the parser-layer __proto__ family in #421 and #425/#428. That fix is what makes this guard safe to add.

pcapkit.protocols.schema.misc.pcapng.Option overrides this method with a namespaced registry of its own and does not delegate here, so it is guarded separately.

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

Parameters:

code (TypeVar(_ET, bound= Enum) | Iterable[TypeVar(_ET, bound= Enum)] | None) – Enumeration code. It can be either a single enumeration or a list of enumerations.
*args (Any) – Arbitrary positional arguments.
**kwargs (Any) – Arbitrary keyword arguments.

If code is provided, the subclass will be registered to the registry mapping with the given code. If code is not given, the subclass will not be registered.

Warns:: pcapkit.utilities.warnings.RegistryWarning – If any of code is already registered, naming the displaced schema and its replacement. This is the same guard register() applies, and it is here as well because a class declaration is the other way into __enum__ – class MyOption(Option, code=...) writes the registry without any call to register(), so guarding only the method would leave the declaration path silently displacing a built-in schema.

Notes

If __enum__ is not yet defined at function call, it will automatically be defined as a _EnumRegistry object, with the default value set to __default__.

If intended to customise the __enum__ mapping, it is possible to override the __init_subclass__() method and define __enum__ manually. Such a manual definition may use a plain collections.defaultdict – e.g. to seed a namespaced or nested mapping such as pcapkit.protocols.schema.misc.pcapng.Option’s – so it is swapped for the retention-safe _EnumRegistry below, before anything else can hold a reference to the original object.

@pcapkit.protocols.schema.schema.schema_final(cls, *, _finalised=True)[source]¶

Finalise schema class.

This decorator function is used to generate necessary attributes and methods for the decorated Schema class. It can be useful to reduce runtime generation time as well as caching already generated attributes.

Notes

The decorator should only be used on the final class. Applying it with _finalised=True seals the class against subclassing, which Schema.__init_subclass__() enforces, and marks it with final() – so @schema_final implies @final and there is never a reason to write both. Writing both is harmless in either order, though; what is not is @final without this decorator, which Schema.__new__() refuses – provided @final is final() (or, on Python 3.11 and up, typing.final() directly). Below 3.11, plain typing.final() does not set __final__ at all (gh-90500), so a class marked with a user’s own from typing import final on 3.10 is not refused: there is nothing on the class for Schema.__new__() to read.

Applying this decorator to the same class a second time only warns: the first application already did the work, so the duplicate is redundant rather than wrong, and the class comes back finalised and usable.

Parameters:

cls (TypeVar(_ST, bound= Type[Schema])) – Schema class.
_finalised (bool) – Whether to make the schema class finalised.

Return type:

TypeVar(_ST, bound= Type[Schema])

Returns:

Finalised schema class.

Warns:

pcapkit.utilities.warnings.SchemaWarning – If cls has already been finalised by this function, i.e. it carries FINAL in its own __dict__. The class is returned untouched.

Internal Definitions¶

class pcapkit.protocols.schema.schema.SchemaMeta(name: str, bases: tuple[type, ...], attrs: dict[str, Any], /, **kwargs: Any)[source]¶

Bases: ABCMeta

Meta class to add dynamic support to Schema.

This meta class is used to generate necessary attributes for the Schema class. It can be useful to reduce runtime generation cost as well as caching already generated attributes.

Schema.__fields__ is a dictionary of field names and their corresponding Field objects, which are used to define and parse the protocol headers. The field dictionary will automatically be populated from the class attributes of the Schema class, and the field names will be the same as the attribute names.

See also

This is implemented thru setting up the initial field dictionary in the __prepare__() method, and then inherit the field dictionaries from the base classes.

Later, during the class creation, the Field.__set_name__ method will be called to set the field name for each field object, as well as to add the field object to the field dictionary.
Schema.__additional__ and Schema.__excluded__ are lists of additional and excluded field names, which are used to determine certain names to be included or excluded from the field dictionary. They will be automatically populated from the class attributes of the Schema class and its base classes.
Note

This is implemented thru the __new__() method, which will inherit the additional and excluded field names from the base classes, as well as populating the additional and excluded field from the subclass attributes.
```
class A(Schema):
    __additional__ = ['a', 'b']

class B(A):
    __additional__ = ['c', 'd']

class C(B):
    __additional__ = ['e', 'f']

print(A.__additional__)  # ['a', 'b']
print(B.__additional__)  # ['a', 'b', 'c', 'd']
print(C.__additional__)  # ['a', 'b', 'c', 'd', 'e', 'f']
```

class pcapkit.protocols.schema.schema.EnumMeta(name: str, bases: tuple[type, ...], attrs: dict[str, Any], /, **kwargs: Any)[source]¶

Bases: SchemaMeta, Generic[_ET]

Meta class to add dynamic support for EnumSchema.

This meta class is used to generate necessary attributes for the SchemaMeta class. It can be useful to reduce runtime generation cost as well as caching already generated attributes.

registry is added to subclasses as an immutable proxy (similar to property, but on class variables) to the EnumSchema.__enum__ mapping.

class pcapkit.protocols.schema.schema._EnumRegistry[source]¶

Bases: defaultdict

A registry collections.defaultdict that never inserts a miss.

EnumSchema.registry (and its class-level twin, EnumMeta.registry) is read with a bare registry[code] at dozens of call sites across the schema layer, e.g. Option.registry[type]. A plain collections.defaultdict inserts whatever default_factory returns the first time an unregistered code is looked up – and since the registry lives on the class, that insertion is permanent and shared by every instance of every subclass in the process. Parsing one packet carrying an unrecognised code is therefore enough to grow the registry for the remainder of the process, and to make a later, entirely legitimate EnumSchema.register() call report an overwrite that never happened.

This is the schema-layer instance of the defect ProtocolBase. _lookup_registry fixed for the protocol-layer __proto__ family in GitHub issues #421 and #425/#428; see GitHub issue #555. The fallback itself is deliberate – it is how an unknown option, chunk or block falls back to its Unknown*/Unassigned* schema – so this subclass keeps returning it, it just stops recording it.

__missing__(key)[source]¶

__missing__(key) # Called by __getitem__ for missing key; pseudo-code: if self.default_factory is None: raise KeyError((key,)) self[key] = value = self.default_factory() return value

Return type:: Any

Type Variables¶

pcapkit.protocols.schema.schema._VT: Any¶

pcapkit.protocols.schema.schema._ET: enum.Enum¶

pcapkit.protocols.schema.schema._ST: Type[pcapkit.protocols.schema.schema.Schema]¶

Data Model¶

class pcapkit.protocols.data.data.Data(dict_=None, **kwargs)[source]¶

Bases: Info

Base class for data models.

__excluded__: list[str]¶: List of names to be excluded from dict conversion.

See also

Please refer to Protocol._decode_next_layer for more information with the inserted names to be excluded.