Transport Layer Base Protocol

pcapkit.protocols.transport.transport contains Transport, which is a base class for transport layer protocols, e.g. TCP, UDP and SCTP.

class pcapkit.protocols.transport.transport.Transport[_PT: Data, _ST: Schema](file=None, length=None, **kwargs)

Bases: ProtocolBase[_PT, _ST], Generic[_PT, _ST]

Abstract base class for transport layer protocol family.

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

Protocol index mapping for decoding next layer, see self._decode_next_layer & self._import_next_layer.

Important

The attribute must be defined and maintained in subclass.

__layer__: Literal['Transport']

Layer of protocol.

property layer: Literal['Transport']

Protocol layer.

classmethod register(code, protocol)

Register a new protocol class.

Notes

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

Parameters:

• code (int) – port number

• protocol (ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]) – module descriptor or a ProtocolBase subclass

Important

Call it on a concrete transport protocol, whose own __proto__ it updates.

Raises:

• pcapkit.utilities.exceptions.UnsupportedCall – If called on Transport itself.

• pcapkit.utilities.exceptions.RegistryError – If protocol is not a class, or not a ProtocolBase subclass.

Warns:

pcapkit.utilities.warnings.RegistryWarning – If this port is already registered and the incumbent differs from the replacement; the message names both. An unresolved ModuleDescriptor counts as different from the class it names. The guard is the one Protocol.register describes (#1504).

Note

Each transport protocol defines its own __proto__, so register_apptype, which calls this once per transport protocol, never warns spuriously.

Changed in version 1.3.0: Takes the protocol as one protocol argument, a ModuleDescriptor or a ProtocolBase subclass, in place of the module and class_ names.

classmethod analyze(ports, payload, **kwargs)

Analyse packet payload.

Parameters:

• ports (tuple[AppType, AppType]) – Source & destination ports, as the application types _make_port() resolves them to – what TCP.src and TCP.dst read off a parsed header, resolved the same way.

• payload (bytes) – Packet payload.

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

Return type:

ProtocolBase

Returns:

Parsed payload as a ProtocolBase instance.

Important

The annotation names the base registry because every transport protocol shares this method while resolving its ports in a registry of its own, such as TCP. A caller that knows its transport gets the narrower type from whatever handed it the pair, e.g. TCP._make_ports.

A bare int port still resolves: a member hashes as its port (AppType.__hash__) and compares equal to it (AppType.__eq__). So each port is tested against __proto__ as it arrived, and the chosen one is unwrapped through AppType.__int__ only when it is a member; unwrapping anything else would make None raise rather than fall back to Raw, and '80' dispatch as a port.

A miss writes nothing back, so an unregistered port does not grow the process-wide registry – see Protocol._lookup_registry.

Changed in version 1.5.0: ports is a pair of AppType members rather than of bare int port numbers, so TCP.submit, through Deferred, hands over the same named ports a parsed header carries (#2417).

classmethod from_data(data, **kwargs)

Create protocol instance from data.

Parameters:

• data (_PT | dict[str, Any]) – Protocol data.

• **kwargs (Any) – Construction keywords, as for Protocol.from_data.

Return type:

Self

Returns:

Protocol instance, cut back to what was captured when the header it was parsed from was cut short (#1458).

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

Post initialisation hook.

Parameters:

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

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

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

Return type:

None

Notes

A header the data ends inside keeps that in its info, so that from_data() rebuilds only what was captured (#1458).

classmethod _make_port(port, proto)

Resolve a port number to its application type.

Parameters:

• port (AppType | int) – port number, or the application type itself

• proto (TransportProtocol) – transport protocol the port belongs to, which is what distinguishes e.g. TCP/80 from UDP/80

Return type:

AppType

Returns:

The AppType for port.

Raises:

ProtocolError – If port is a bool (#2287), or is neither an int that fits in 16 bits nor an AppType.

Important

The schema field converts a bare int port only on the way out (PortEnumField.pre_process), so normalising here gives a constructed packet the same AppType as a parsed one.

Changed in version 1.6.0: A port past 16 bits, a negative one, or one that is not an int raises ProtocolError, rather than a bare EnumValueError (#2594).

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

Decode next layer protocol.

The next layer is looked up by port number. The lower of the two ports is the primary key; the higher one is used only when it alone is registered.

Parameters:

• dict_ (_PT) – info buffer

• ports (tuple[int, int]) – source & destination port numbers

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

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

Return type:

_PT

Returns:

Current protocol with next layer extracted.

Important

The port is forwarded whether or not it is registered: Protocol._import_next_layer passes it on as alias and Raw records it as Data_Raw.protocol, which tells payloads on different undecoded ports apart. The port carried is the one looked up. SCTP._decode_next_layer and Internet._import_next_layer behave the same way for an unregistered PPID and transport type.