diff --git a/docs/source/pcapkit/corekit/fields/index.rst b/docs/source/pcapkit/corekit/fields/index.rst index b888dcbb40..5626ea7bd6 100644 --- a/docs/source/pcapkit/corekit/fields/index.rst +++ b/docs/source/pcapkit/corekit/fields/index.rst @@ -34,6 +34,7 @@ hierarchy: NumberField --> Int16Field & UInt16Field NumberField --> Int8Field & UInt8Field NumberField --> EnumField + EnumField --> PortEnumField end C --> NumberField @@ -87,6 +88,7 @@ hierarchy: click Int8Field "/pcapkit/corekit/fields/numbers.html#pcapkit.corekit.fields.numbers.Int8Field" click UInt8Field "/pcapkit/corekit/fields/numbers.html#pcapkit.corekit.fields.numbers.UInt8Field" click EnumField "/pcapkit/corekit/fields/numbers.html#pcapkit.corekit.fields.numbers.EnumField" + click PortEnumField "/pcapkit/corekit/fields/numbers.html#pcapkit.corekit.fields.numbers.PortEnumField" click _TextField "/pcapkit/corekit/fields/strings.html#pcapkit.corekit.fields.strings._TextField" click BytesField "/pcapkit/corekit/fields/strings.html#pcapkit.corekit.fields.strings.BytesField" diff --git a/docs/source/pcapkit/corekit/fields/numbers.rst b/docs/source/pcapkit/corekit/fields/numbers.rst index bc078341b2..8a21fe1ce8 100644 --- a/docs/source/pcapkit/corekit/fields/numbers.rst +++ b/docs/source/pcapkit/corekit/fields/numbers.rst @@ -45,6 +45,10 @@ Enumeration Fields :members: :show-inheritance: +.. autoclass:: pcapkit.corekit.fields.numbers.PortEnumField + :members: + :show-inheritance: + Internal Definitions -------------------- diff --git a/pcapkit/corekit/fields/numbers.py b/pcapkit/corekit/fields/numbers.py index 2d285275c5..5781618a29 100644 --- a/pcapkit/corekit/fields/numbers.py +++ b/pcapkit/corekit/fields/numbers.py @@ -17,7 +17,7 @@ 'Int16Field', 'UInt16Field', 'Int64Field', 'UInt64Field', 'Int8Field', 'UInt8Field', - 'EnumField', + 'EnumField', 'PortEnumField', ] if TYPE_CHECKING: @@ -27,6 +27,7 @@ from aenum import IntEnum as AenumEnum from typing_extensions import Literal, Self + from pcapkit.const.reg.apptype import AppType, TransportProtocol from pcapkit.corekit.fields.field import NoValueType _T = TypeVar('_T', bound='int') @@ -822,8 +823,10 @@ def _unregistered_member(namespace: 'Type[StdlibEnum] | Type[AenumEnum]', Note: Building a member this way, rather than as some other type - altogether, is why :meth:`~pcapkit.protocols.schema.transport.tcp.PortEnumField.post_process` - and its siblings need this rather than :meth:`_pseudo_member`: + altogether, is why :meth:`PortEnumField.post_process` and + :meth:`OptionEnumField.post_process + ` + need this rather than :meth:`_pseudo_member`: the result answers ``isinstance(result, AppType)`` truthfully, which matters to at least seven ``isinstance`` sites elsewhere in :mod:`pcapkit.protocols` (see ``test_a_member_is_still_an_apptype`` @@ -860,7 +863,7 @@ def _unregistered_member(namespace: 'Type[StdlibEnum] | Type[AenumEnum]', one lookup this member is deliberately absent from. A minted member is registered and so round-trips; left alone, this one would not. Measured on CPython 3.14.7, resolving port 53406 - through :class:`~pcapkit.protocols.schema.transport.tcp.PortEnumField` + through the TCP schema's :class:`PortEnumField` with nothing in place of the override, ``pickle.loads(pickle.dumps(member))`` raised ``ValueError: 'unknown [53406 - tcp]' is not a valid TCP`` -- while @@ -996,3 +999,147 @@ def _reduce_unregistered_member( # pylint: disable=unused-argument """ return (_rebuild_unregistered_member, (namespace, value, name, attrs)) + + +class PortEnumField(EnumField): + """Enumerated value for transport-layer port fields. + + Args: + length: Field size (in bytes); if a callable is given, it should return + an integer value and accept the current packet as its only argument. + default: Field default value, if any. + signed: Whether the field is signed; :data:`None` defers to the + class-level ``__signed__``, which this class leaves unset and so + means unsigned. + byteorder: Field byte order. + bit_length: Field bit length. + namespace: The port registry, i.e. + :class:`~pcapkit.const.reg.apptype.AppType`. + callback: Callback function to be called upon + :meth:`self.__call__ `. + proto: The transport protocol whose ports this field carries -- a single + :class:`~pcapkit.const.reg.apptype.TransportProtocol` member, which + is what tells e.g. TCP/80 from UDP/80. The TCP, UDP and SCTP + schemas each pass their own. + + Important: + This class is specifically designed for :class:`~pcapkit.const.reg.apptype.AppType` + as it is actually a :class:`~enum.StrEnum` class. + + """ + if TYPE_CHECKING: + _namespace: 'Type[AppType]' + + def __init__(self, length: 'int | Callable[[dict[str, Any]], int]', + default: 'StdlibEnum | AenumEnum | NoValueType' = NO_VALUE, + signed: 'Optional[bool]' = None, + byteorder: 'Literal["little", "big"]' = 'big', + bit_length: 'Optional[int]' = None, + namespace: 'Optional[Type[AppType]]' = None, + callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None, *, + proto: 'TransportProtocol') -> 'None': + super().__init__(length, default, signed, byteorder, bit_length, namespace, callback) + + self._proto = proto + + def pre_process(self, value: 'int | AppType', packet: 'dict[str, Any]') -> 'int | bytes': + """Process field value before construction (packing). + + Arguments: + value: Field value. + packet: Packet data. + + Returns: + Processed field value. + + """ + # NOTE: imported here rather than at module level, as + # :mod:`pcapkit.const` itself imports :mod:`pcapkit.corekit`; and tested + # against AppType rather than ``self._namespace``, as the three + # per-transport copies this class replaced did (:issue:`1516`). + from pcapkit.const.reg.apptype import AppType # pylint: disable=import-outside-toplevel + + if isinstance(value, AppType): + value = value.port + return super().pre_process(value, packet) + + def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'AppType': + """Process field value after parsing (unpacked). + + Args: + value: Field value. + packet: Packet data. + + Returns: + Processed field value -- the registry member declared for the + port, or an unregistered member of the same registry, carrying + the port itself, when the registry declares none. + + Notes: + :meth:`~pcapkit.const.reg.apptype.AppType.get` mints a fresh + member -- via :func:`aenum.extend_enum` -- for any port neither an + existing row nor one of :meth:`_missing_`'s documented IANA spans + accounts for, which in practice means the ephemeral/dynamic range. + Calling it on every parsed port would therefore grow the registry + without bound. This peeks at the registry + :meth:`~pcapkit.const.reg.apptype.AppType.get` itself would + consult -- its per-port rows via ``__registry__.getlist``, then + its documented spans via ``_missing_`` -- and only calls + :meth:`~pcapkit.const.reg.apptype.AppType.get` once one of those is + already known to hold, so a genuine miss gets + :meth:`EnumField._unregistered_member` instead of a mint. + + A port outside this field's own width is rejected *before* any of + that, rather than being let through to :meth:`_missing_` and + caught alongside a genuine miss. Both are a bare :exc:`ValueError` + with nothing to tell them apart by type, and + :meth:`~pcapkit.const.reg.apptype.AppType.get` rejects the + out-of-range case deliberately, so that it is never minted over -- a + catch keyed on exception type alone cannot see the difference + between that and :mod:`aenum`'s own "no member has this value", + so it would absorb both and quietly turn the rejection into a + miss. Checking the width first needs no exception-based + distinction at all. ``self.length`` is the field's own declared byte + width (``2`` for every caller of this class, hence ``0``-``65535``) + rather than a hard-coded ``65535`` borrowed from + :meth:`~pcapkit.const.reg.apptype.AppType._missing_`'s own guard, + so the two stay in lockstep by construction. + + """ + value = super(EnumField, self).post_process(value, packet) + proto = self._proto + if not (isinstance(value, int) and 0 <= value < (1 << (8 * self.length))): + # NOTE: lets AppType.get() -- unmodified -- raise its rejection + # for a port this field's own width cannot represent, rather than + # risking it being absorbed below as a foreign miss. + return self._namespace.get(value, proto=proto) + owner = self._namespace._dispatch(value, proto) # pylint: disable=protected-access + if not owner.__registry__.getlist(value): # type: ignore[union-attr] + try: + declared = owner._missing_(value) # pylint: disable=protected-access + except ValueError as error: + # NOTE: value is already known to be in-width here, so this + # ValueError is aenum's own "no member has this value" for an + # in-range but unassigned port -- a foreign miss, absorbed -- + # never the out-of-range rejection, which never reaches + # this branch. A pcapkit.utilities.exceptions error is still a + # deliberate registry decision and propagates unchanged. + if isinstance(error, BaseError): + raise + declared = None + if declared is None: + # NOTE: an unregistered member of ``owner`` itself, per the + # ruling in EnumField._unregistered_member, rather than a foreign + # pseudo-enum. ``.port``, ``.svc`` and ``.proto`` are what a real + # AppType member carries -- read unconditionally by e.g. + # Transport._decode_next_layer's ``srcport.port`` -- and + # ``svc='unknown'`` marks a port with no declared service. They + # are passed in the order AppType.__new__ sets them, because + # DictDumper.object_hook renders a member's addon keys straight + # out of its ``__dict__`` in insertion order -- so any other + # order here would make an unassigned port dump ``port`` before + # ``svc`` while every declared one dumps ``svc`` first. + return self._unregistered_member( + owner, f'unknown [{value:d} - {proto.name}]', + svc='unknown', port=value, proto=proto) + return self._namespace.get(value, proto=proto) diff --git a/pcapkit/protocols/schema/transport/sctp.py b/pcapkit/protocols/schema/transport/sctp.py index b2143da213..2366f1f3ef 100644 --- a/pcapkit/protocols/schema/transport/sctp.py +++ b/pcapkit/protocols/schema/transport/sctp.py @@ -15,10 +15,9 @@ from pcapkit.corekit.fields.collections import ListField, OptionField from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField from pcapkit.corekit.fields.misc import PayloadField, SchemaField -from pcapkit.corekit.fields.numbers import EnumField, UInt16Field, UInt32Field +from pcapkit.corekit.fields.numbers import EnumField, PortEnumField, UInt16Field, UInt32Field from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final -from pcapkit.utilities.exceptions import BaseError from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING __all__ = [ @@ -209,78 +208,6 @@ def callback(pkt: 'dict[str, Any]') -> 'int': return callback -class PortEnumField(EnumField): - """Enumerated value for protocol fields. - - Args: - length: Field size (in bytes); if a callable is given, it should return - an integer value and accept the current packet as its only argument. - default: Field default value, if any. - signed: Whether the field is signed. - byteorder: Field byte order. - bit_length: Field bit length. - callback: Callback function to be called upon - :meth:`self.__call__ `. - - Important: - This class is specifically designed for :class:`~pcapkit.const.reg.apptype.AppType` - as it is actually a :class:`~enum.StrEnum` class. - - """ - if TYPE_CHECKING: - _namespace: 'Enum_AppType' - - def pre_process(self, value: 'int | Enum_AppType', packet: 'dict[str, Any]') -> 'int | bytes': - """Process field value before construction (packing). - - Arguments: - value: Field value. - packet: Packet data. - - Returns: - Processed field value. - - """ - if isinstance(value, Enum_AppType): - value = value.port - return super().pre_process(value, packet) - - def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'Enum_AppType': - """Process field value after parsing (unpacked). - - Args: - value: Field value. - packet: Packet data. - - Returns: - Processed field value -- the registry member declared for the - port, or an unregistered member of the same registry, carrying - the port itself, when the registry declares none. - - Notes: - See :meth:`pcapkit.protocols.schema.transport.tcp.PortEnumField.post_process`, - whose notes this mirrors verbatim aside from the transport. - - """ - value = super(EnumField, self).post_process(value, packet) - proto = Enum_TransportProtocol.sctp - if not (isinstance(value, int) and 0 <= value < (1 << (8 * self.length))): - return self._namespace.get(value, proto=proto) - owner = self._namespace._dispatch(value, proto) # pylint: disable=protected-access - if not owner.__registry__.getlist(value): # type: ignore[union-attr] - try: - declared = owner._missing_(value) # pylint: disable=protected-access - except ValueError as error: - if isinstance(error, BaseError): - raise - declared = None - if declared is None: - return self._unregistered_member( - owner, f'unknown [{value:d} - {proto.name}]', - svc='unknown', port=value, proto=proto) - return self._namespace.get(value, proto=proto) - - class ErrorCause(EnumSchema[Enum_CauseCode]): """Header schema for SCTP error causes.""" @@ -1013,9 +940,11 @@ class SCTP(Schema): """ #: Source port. - srcport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType) + srcport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType, + proto=Enum_TransportProtocol.sctp) #: Destination port. - dstport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType) + dstport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType, + proto=Enum_TransportProtocol.sctp) #: Verification tag. vtag: 'int' = UInt32Field() #: Checksum, as a CRC32c over the whole packet with this field zeroed. diff --git a/pcapkit/protocols/schema/transport/tcp.py b/pcapkit/protocols/schema/transport/tcp.py index 576d0a7cbe..2a89a227fa 100644 --- a/pcapkit/protocols/schema/transport/tcp.py +++ b/pcapkit/protocols/schema/transport/tcp.py @@ -14,11 +14,11 @@ from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, NoValueField, PayloadField, SchemaField, SwitchField) -from pcapkit.corekit.fields.numbers import (EnumField, UInt8Field, UInt16Field, UInt32Field, - UInt64Field) +from pcapkit.corekit.fields.numbers import (EnumField, PortEnumField, UInt8Field, UInt16Field, + UInt32Field, UInt64Field) from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final -from pcapkit.utilities.exceptions import BaseError, FieldError +from pcapkit.utilities.exceptions import FieldError from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING __all__ = [ @@ -369,124 +369,6 @@ def mptcp_dss_dsn_selector(pkt: 'dict[str, Any]') -> 'Field': return UInt64Field() if pkt['flags']['m'] else UInt32Field() -class PortEnumField(EnumField): - """Enumerated value for protocol fields. - - Args: - length: Field size (in bytes); if a callable is given, it should return - an integer value and accept the current packet as its only argument. - default: Field default value, if any. - signed: Whether the field is signed. - byteorder: Field byte order. - bit_length: Field bit length. - callback: Callback function to be called upon - :meth:`self.__call__ `. - - Important: - This class is specifically designed for :class:`~pcapkit.const.reg.apptype.AppType` - as it is actually a :class:`~enum.StrEnum` class. - - """ - if TYPE_CHECKING: - _namespace: 'Enum_AppType' - - def pre_process(self, value: 'int | Enum_AppType', packet: 'dict[str, Any]') -> 'int | bytes': - """Process field value before construction (packing). - - Arguments: - value: Field value. - packet: Packet data. - - Returns: - Processed field value. - - """ - if isinstance(value, Enum_AppType): - value = value.port - return super().pre_process(value, packet) - - def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'Enum_AppType': - """Process field value after parsing (unpacked). - - Args: - value: Field value. - packet: Packet data. - - Returns: - Processed field value -- the registry member declared for the - port, or an unregistered member of the same registry, carrying - the port itself, when the registry declares none. - - Notes: - :meth:`~pcapkit.const.reg.apptype.AppType.get` mints a fresh - member -- via :func:`aenum.extend_enum` -- for any port neither an - existing row nor one of :meth:`_missing_`'s documented IANA spans - accounts for, which in practice means the ephemeral/dynamic range. - Calling it on every parsed port would therefore grow the registry - without bound. This peeks at the registry - :meth:`~pcapkit.const.reg.apptype.AppType.get` itself would - consult -- its per-port rows via ``__registry__.getlist``, then - its documented spans via ``_missing_`` -- and only calls - :meth:`~pcapkit.const.reg.apptype.AppType.get` once one of those is - already known to hold, so a genuine miss gets - :meth:`EnumField._unregistered_member` instead of a mint. - - A port outside this field's own width is rejected *before* any of - that, rather than being let through to :meth:`_missing_` and - caught alongside a genuine miss. Both are a bare :exc:`ValueError` - with nothing to tell them apart by type, and - :meth:`~pcapkit.const.reg.apptype.AppType.get` rejects the - out-of-range case deliberately, so that it is never minted over -- a - catch keyed on exception type alone cannot see the difference - between that and :mod:`aenum`'s own "no member has this value", - so it would absorb both and quietly turn the rejection into a - miss. Checking the width first needs no exception-based - distinction at all. ``self.length`` is the field's own declared byte - width (``2`` for every caller of this class, hence ``0``-``65535``) - rather than a hard-coded ``65535`` borrowed from - :meth:`~pcapkit.const.reg.apptype.AppType._missing_`'s own guard, - so the two stay in lockstep by construction. - - """ - value = super(EnumField, self).post_process(value, packet) - proto = Enum_TransportProtocol.tcp - if not (isinstance(value, int) and 0 <= value < (1 << (8 * self.length))): - # NOTE: lets AppType.get() -- unmodified -- raise its rejection - # for a port this field's own width cannot represent, rather than - # risking it being absorbed below as a foreign miss. - return self._namespace.get(value, proto=proto) - owner = self._namespace._dispatch(value, proto) # pylint: disable=protected-access - if not owner.__registry__.getlist(value): # type: ignore[union-attr] - try: - declared = owner._missing_(value) # pylint: disable=protected-access - except ValueError as error: - # NOTE: value is already known to be in-width here, so this - # ValueError is aenum's own "no member has this value" for an - # in-range but unassigned port -- a foreign miss, absorbed -- - # never the out-of-range rejection, which never reaches - # this branch. A pcapkit.utilities.exceptions error is still a - # deliberate registry decision and propagates unchanged. - if isinstance(error, BaseError): - raise - declared = None - if declared is None: - # NOTE: an unregistered member of ``owner`` itself, per the - # ruling in EnumField._unregistered_member, rather than a foreign - # pseudo-enum. ``.port``, ``.svc`` and ``.proto`` are what a real - # AppType member carries -- read unconditionally by e.g. - # Transport._decode_next_layer's ``srcport.port`` -- and - # ``svc='unknown'`` marks a port with no declared service. They - # are passed in the order AppType.__new__ sets them, because - # DictDumper.object_hook renders a member's addon keys straight - # out of its ``__dict__`` in insertion order -- so any other - # order here would make an unassigned port dump ``port`` before - # ``svc`` while every declared one dumps ``svc`` first. - return self._unregistered_member( - owner, f'unknown [{value:d} - {proto.name}]', - svc='unknown', port=value, proto=proto) - return self._namespace.get(value, proto=proto) - - class Option(EnumSchema[Enum_Option]): """Header schema for TCP options.""" @@ -1177,9 +1059,11 @@ class TCP(Schema): """Header schema for TCP packet.""" #: Source port. - srcport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType) + srcport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType, + proto=Enum_TransportProtocol.tcp) #: Destination port. - dstport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType) + dstport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType, + proto=Enum_TransportProtocol.tcp) #: Sequence number. seq: 'int' = UInt32Field() #: Acknowledgement number. diff --git a/pcapkit/protocols/schema/transport/udp.py b/pcapkit/protocols/schema/transport/udp.py index 10b1d27c88..8df891fb8d 100644 --- a/pcapkit/protocols/schema/transport/udp.py +++ b/pcapkit/protocols/schema/transport/udp.py @@ -7,99 +7,26 @@ from pcapkit.const.reg.apptype import AppType as Enum_AppType from pcapkit.const.reg.apptype import TransportProtocol as Enum_TransportProtocol from pcapkit.corekit.fields.misc import PayloadField -from pcapkit.corekit.fields.numbers import EnumField, UInt16Field +from pcapkit.corekit.fields.numbers import PortEnumField, UInt16Field from pcapkit.corekit.fields.strings import BytesField from pcapkit.protocols.schema.schema import Schema, schema_final -from pcapkit.utilities.exceptions import BaseError __all__ = ['UDP'] if TYPE_CHECKING: - from typing import Any - from pcapkit.protocols.protocol import ProtocolBase -class PortEnumField(EnumField): - """Enumerated value for protocol fields. - - Args: - length: Field size (in bytes); if a callable is given, it should return - an integer value and accept the current packet as its only argument. - default: Field default value, if any. - signed: Whether the field is signed. - byteorder: Field byte order. - bit_length: Field bit length. - callback: Callback function to be called upon - :meth:`self.__call__ `. - - Important: - This class is specifically designed for :class:`~pcapkit.const.reg.apptype.AppType` - as it is actually a :class:`~enum.StrEnum` class. - - """ - if TYPE_CHECKING: - _namespace: 'Enum_AppType' - - def pre_process(self, value: 'int | Enum_AppType', packet: 'dict[str, Any]') -> 'int | bytes': - """Process field value before construction (packing). - - Arguments: - value: Field value. - packet: Packet data. - - Returns: - Processed field value. - - """ - if isinstance(value, Enum_AppType): - value = value.port - return super().pre_process(value, packet) - - def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'Enum_AppType': - """Process field value after parsing (unpacked). - - Args: - value: Field value. - packet: Packet data. - - Returns: - Processed field value -- the registry member declared for the - port, or an unregistered member of the same registry, carrying - the port itself, when the registry declares none. - - Notes: - See :meth:`pcapkit.protocols.schema.transport.tcp.PortEnumField.post_process`, - whose notes this mirrors verbatim aside from the transport. - - """ - value = super(EnumField, self).post_process(value, packet) - proto = Enum_TransportProtocol.udp - if not (isinstance(value, int) and 0 <= value < (1 << (8 * self.length))): - return self._namespace.get(value, proto=proto) - owner = self._namespace._dispatch(value, proto) # pylint: disable=protected-access - if not owner.__registry__.getlist(value): # type: ignore[union-attr] - try: - declared = owner._missing_(value) # pylint: disable=protected-access - except ValueError as error: - if isinstance(error, BaseError): - raise - declared = None - if declared is None: - return self._unregistered_member( - owner, f'unknown [{value:d} - {proto.name}]', - svc='unknown', port=value, proto=proto) - return self._namespace.get(value, proto=proto) - - @schema_final class UDP(Schema): """Header schema for UDP packet.""" #: Source port. - srcport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType) + srcport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType, + proto=Enum_TransportProtocol.udp) #: Destination port. - dstport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType) + dstport: 'Enum_AppType' = PortEnumField(length=2, namespace=Enum_AppType, + proto=Enum_TransportProtocol.udp) #: Length of UDP packet. len: 'int' = UInt16Field() #: Checksum of UDP packet. diff --git a/pcapkit/protocols/transport/transport.py b/pcapkit/protocols/transport/transport.py index 299bb4d76d..c267c1dbe4 100644 --- a/pcapkit/protocols/transport/transport.py +++ b/pcapkit/protocols/transport/transport.py @@ -221,7 +221,7 @@ def _make_port(port: 'Enum_AppType | int', :meth:`self.make ` accepts a bare :obj:`int` for a port, but the schema field converts it only on the way *out* (in :meth:`PortEnumField.pre_process - `), + `), so the schema attribute keeps whatever it was handed. Normalising here makes a constructed packet carry the same :class:`~pcapkit.const.reg.apptype.AppType` as a parsed one, so diff --git a/tests/const/test_const_apptype_split_unit.py b/tests/const/test_const_apptype_split_unit.py index 33b757a39c..c516140ba3 100644 --- a/tests/const/test_const_apptype_split_unit.py +++ b/tests/const/test_const_apptype_split_unit.py @@ -1018,14 +1018,16 @@ def test_no_lookup_call_site_in_the_library_builds_a_composite_proto(self) -> No """Why refusing one breaks nothing, asserted rather than taken on trust. The refusal is only safe while every library caller passes a single - transport protocol. There are ten call sites into the base registry's lookup + transport protocol. There are four call sites into the base registry's lookup outside :mod:`pcapkit.const`: one in ``Transport._make_port``, which passes the ``proto`` its six call sites hand it -- a literal member every time, two - per transport in the TCP, UDP and SCTP protocol modules -- and three in each - of the TCP, UDP and SCTP schemas' ``PortEnumField.post_process``, which - binds ``proto`` to one literal member and then uses it for ``_dispatch`` and - both ``get`` calls. None of them can pass a composite, because none of those - modules builds one. + per transport in the TCP, UDP and SCTP protocol modules -- and three in the + shared ``PortEnumField.post_process`` in :mod:`pcapkit.corekit.fields.numbers`, + which binds ``proto`` to the member its constructor was handed and then uses + it for ``_dispatch`` and both ``get`` calls. That constructor is called six + times, two per transport in the TCP, UDP and SCTP schemas, each with a + literal member (GitHub issue #1516). None of them can pass a composite, + because none of those modules builds one. Read off disk rather than through :func:`inspect.getsource`, so this needs no import of :mod:`pcapkit.protocols` -- whose runtime dependencies this @@ -1065,6 +1067,7 @@ def test_no_lookup_call_site_in_the_library_builds_a_composite_proto(self) -> No composites = [] # type: list[tuple[str, int]] lookups = [] # type: list[str] bindings = [] # type: list[tuple[str, str]] + stored = [] # type: list[tuple[str, str]] for relative in callers: path = root / relative with self.subTest(module=relative): @@ -1075,20 +1078,24 @@ def test_no_lookup_call_site_in_the_library_builds_a_composite_proto(self) -> No if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr): if 'TransportProtocol' in ast.unparse(node): composites.append((relative, node.lineno)) - # What ``proto`` is bound to, for the schema fields that use a - # local rather than passing the member straight in. Only those - # three modules: ``Transport._decode_next_layer`` has a local of - # the same name holding a *port number* for the next-layer - # registry, which never reaches this lookup. - if isinstance(node, ast.Assign) and 'schema' in relative: + # What ``proto`` is bound to, for the port field that uses a + # local rather than passing the member straight in, and what + # that field stores the member it was constructed with as. Not + # the transport modules: ``Transport._decode_next_layer`` has a + # local of the same name holding a *port number* for the + # next-layer registry, which never reaches this lookup. + if isinstance(node, ast.Assign) and ('schema' in relative + or relative.endswith('numbers.py')): for target in node.targets: if isinstance(target, ast.Name) and target.id == 'proto': bindings.append((relative, ast.unparse(node.value))) + if isinstance(target, ast.Attribute) and target.attr == '_proto': + stored.append((relative, ast.unparse(node.value))) if not isinstance(node, ast.Call): continue name = (node.func.attr if isinstance(node.func, ast.Attribute) else node.func.id if isinstance(node.func, ast.Name) else '') - if name not in ('get', 'get_all', '_dispatch', '_make_port'): + if name not in ('get', 'get_all', '_dispatch', '_make_port', 'PortEnumField'): continue arguments = [keyword.value for keyword in node.keywords if keyword.arg == 'proto'] @@ -1111,27 +1118,27 @@ def test_no_lookup_call_site_in_the_library_builds_a_composite_proto(self) -> No with self.subTest(call=rendered): self.assertIn(rendered.split('proto=', 1)[1].rstrip(')'), permitted) - # Every local named ``proto`` in these modules is bound to exactly one - # member, which is what makes the bare ``proto`` arguments above single-bit. - self.assertEqual(sorted(bindings), [ - ('pcapkit/protocols/schema/transport/sctp.py', 'Enum_TransportProtocol.sctp'), - ('pcapkit/protocols/schema/transport/tcp.py', 'Enum_TransportProtocol.tcp'), - ('pcapkit/protocols/schema/transport/udp.py', 'Enum_TransportProtocol.udp'), - ]) + # The port field's local ``proto`` is the member its constructor stored, + # and the constructor stores nothing but its own ``proto`` argument -- which + # every call site below passes as one literal member. That chain is what + # makes the bare ``proto`` arguments above single-bit. + self.assertEqual(bindings, [('pcapkit/corekit/fields/numbers.py', 'self._proto')]) + self.assertEqual(stored, [('pcapkit/corekit/fields/numbers.py', 'proto')]) # The sixteen the sweep found, so a new one cannot appear without this test # saying so. ``transport.py`` holds the single lookup and the six literal - # ``_make_port`` arguments that feed it; each schema module holds three. + # ``_make_port`` arguments that feed it; ``numbers.py`` holds the port + # field's three, and each schema module constructs that field twice. self.assertEqual(sorted(lookups), [ - 'pcapkit/protocols/schema/transport/sctp.py _dispatch(proto=proto)', - 'pcapkit/protocols/schema/transport/sctp.py get(proto=proto)', - 'pcapkit/protocols/schema/transport/sctp.py get(proto=proto)', - 'pcapkit/protocols/schema/transport/tcp.py _dispatch(proto=proto)', - 'pcapkit/protocols/schema/transport/tcp.py get(proto=proto)', - 'pcapkit/protocols/schema/transport/tcp.py get(proto=proto)', - 'pcapkit/protocols/schema/transport/udp.py _dispatch(proto=proto)', - 'pcapkit/protocols/schema/transport/udp.py get(proto=proto)', - 'pcapkit/protocols/schema/transport/udp.py get(proto=proto)', + 'pcapkit/corekit/fields/numbers.py _dispatch(proto=proto)', + 'pcapkit/corekit/fields/numbers.py get(proto=proto)', + 'pcapkit/corekit/fields/numbers.py get(proto=proto)', + 'pcapkit/protocols/schema/transport/sctp.py PortEnumField(proto=Enum_TransportProtocol.sctp)', + 'pcapkit/protocols/schema/transport/sctp.py PortEnumField(proto=Enum_TransportProtocol.sctp)', + 'pcapkit/protocols/schema/transport/tcp.py PortEnumField(proto=Enum_TransportProtocol.tcp)', + 'pcapkit/protocols/schema/transport/tcp.py PortEnumField(proto=Enum_TransportProtocol.tcp)', + 'pcapkit/protocols/schema/transport/udp.py PortEnumField(proto=Enum_TransportProtocol.udp)', + 'pcapkit/protocols/schema/transport/udp.py PortEnumField(proto=Enum_TransportProtocol.udp)', 'pcapkit/protocols/transport/sctp.py _make_port(proto=Enum_TransportProtocol.sctp)', 'pcapkit/protocols/transport/sctp.py _make_port(proto=Enum_TransportProtocol.sctp)', 'pcapkit/protocols/transport/tcp.py _make_port(proto=Enum_TransportProtocol.tcp)', diff --git a/tests/corekit/test_field_roundtrip_unit.py b/tests/corekit/test_field_roundtrip_unit.py index bf1dd669c0..366ae210e2 100644 --- a/tests/corekit/test_field_roundtrip_unit.py +++ b/tests/corekit/test_field_roundtrip_unit.py @@ -116,6 +116,7 @@ def field_cases() -> 'tuple[FieldCase, ...]': # pylint: disable=too-many-locals """Every field case, built against the current :mod:`pcapkit` import.""" import aenum + from pcapkit.const.reg.apptype import AppType, TransportProtocol from pcapkit.const.reg.transtype import TransType from pcapkit.corekit.fields import collections as fcoll from pcapkit.corekit.fields import ipaddress as fip @@ -178,6 +179,13 @@ class Aenum(aenum.IntEnum): cases.append(FieldCase( 'EnumField/bit-length-3', lambda: fnum.EnumField(length=1, bit_length=3, namespace=None), (0, 7), (b'\x00', b'\x07', b'\xff'))) + # Port 80 is declared on all three transports, so building the value mints + # nothing; 54321 is declared on none, and resolves unregistered (#575). + for proto in (TransportProtocol.tcp, TransportProtocol.udp, TransportProtocol.sctp): + cases.append(FieldCase( + f'PortEnumField/{proto.name}', + lambda proto=proto: fnum.PortEnumField(length=2, namespace=AppType, proto=proto), + (AppType.get(80, proto=proto),), _wires(2) + (b'\xd4\x31',))) # -- strings ------------------------------------------------------------ for width in (0, 1, 4): diff --git a/tests/corekit/test_fields_numbers_port_enum_field_1516_unit.py b/tests/corekit/test_fields_numbers_port_enum_field_1516_unit.py new file mode 100644 index 0000000000..0d5df7a594 --- /dev/null +++ b/tests/corekit/test_fields_numbers_port_enum_field_1516_unit.py @@ -0,0 +1,127 @@ +# -*- coding: utf-8 -*- +"""One :class:`~pcapkit.corekit.fields.numbers.PortEnumField` for TCP, UDP and SCTP. + +GitHub issue #1516. The TCP, UDP and SCTP schemas each defined their own +``PortEnumField``, three copies that differed only in the +:class:`~pcapkit.const.reg.apptype.TransportProtocol` member hard-coded into +``post_process``. The owner ruled that they become one field in +:mod:`pcapkit.corekit.fields.numbers`, beside +:class:`~pcapkit.corekit.fields.numbers.EnumField`, with a ``proto`` parameter +each schema fills in with its own transport. + +These tests pin the shared field: that each schema uses it with its own +transport, that ``proto`` alone decides which registry a port resolves in, that +an unassigned port resolves without growing the registry, and that packing and +unpacking close. + +""" +import unittest +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + + +class PortEnumFieldTests(unittest.TestCase): + """The shared port field, over each transport it serves.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _field(self, proto: 'Any') -> 'Any': + from pcapkit.const.reg.apptype import AppType + from pcapkit.corekit.fields.numbers import PortEnumField + + return PortEnumField(length=2, namespace=AppType, proto=proto) + + def _protos(self) -> 'tuple[Any, ...]': + from pcapkit.const.reg.apptype import TransportProtocol + + return (TransportProtocol.tcp, TransportProtocol.udp, TransportProtocol.sctp) + + def test_it_is_public_beside_enum_field(self) -> None: + from pcapkit.corekit.fields import numbers + + self.assertIn('PortEnumField', numbers.__all__) + self.assertTrue(issubclass(numbers.PortEnumField, numbers.EnumField)) + + def test_each_schema_uses_the_shared_field_with_its_own_transport(self) -> None: + from pcapkit.const.reg.apptype import TransportProtocol + from pcapkit.corekit.fields.numbers import PortEnumField + from pcapkit.protocols.schema.transport.sctp import SCTP + from pcapkit.protocols.schema.transport.tcp import TCP + from pcapkit.protocols.schema.transport.udp import UDP + + for schema, proto in ((TCP, TransportProtocol.tcp), (UDP, TransportProtocol.udp), + (SCTP, TransportProtocol.sctp)): + for name in ('srcport', 'dstport'): + with self.subTest(schema=schema.__name__, field=name): + field = schema.__fields__[name] + self.assertIs(type(field), PortEnumField) + self.assertIs(field._proto, proto) # pylint: disable=protected-access + + def test_proto_is_required(self) -> None: + from pcapkit.const.reg.apptype import AppType + from pcapkit.corekit.fields.numbers import PortEnumField + + with self.assertRaises(TypeError): + PortEnumField(length=2, namespace=AppType) # type: ignore[call-arg] # pylint: disable=missing-kwoa + + def test_proto_decides_the_registry_a_declared_port_resolves_in(self) -> None: + """The same two octets, three transports, three answers. + + Port 512 is ``exec`` on TCP and ``biff`` on UDP, so a field that ignored + ``proto`` could not get both right. + + """ + from pcapkit.const.reg.apptype import AppType, TransportProtocol + + for proto, port, octets, svc in ((TransportProtocol.tcp, 512, b'\x02\x00', 'exec'), + (TransportProtocol.udp, 512, b'\x02\x00', 'biff'), + (TransportProtocol.tcp, 80, b'\x00\x50', 'http'), + (TransportProtocol.udp, 53, b'\x00\x35', 'domain'), + (TransportProtocol.sctp, 9, b'\x00\x09', 'discard')): + with self.subTest(proto=proto.name, port=port): + resolved = self._field(proto).unpack(octets, {}) + self.assertIs(resolved, AppType.get(port, proto=proto)) + self.assertIsInstance(resolved, AppType.__registries__[proto]) + self.assertEqual((resolved.svc, resolved.port, resolved.proto), (svc, port, proto)) + + def test_an_unknown_port_resolves_unregistered_in_its_own_transport(self) -> None: + from pcapkit.const.reg.apptype import AppType + + for proto in self._protos(): + with self.subTest(proto=proto.name): + owner = AppType.__registries__[proto] + before = len(owner.__members__) + + resolved = self._field(proto).unpack(b'\xd4\x31', {}) # 54321 + + self.assertIsInstance(resolved, owner) + self.assertEqual(resolved.name, '') + self.assertEqual(resolved.value, f'unknown [54321 - {proto.name}]') + self.assertEqual((resolved.svc, resolved.port, resolved.proto), + ('unknown', 54321, proto)) + self.assertEqual(len(owner.__members__), before) + + def test_pack_takes_a_member_or_a_bare_port(self) -> None: + from pcapkit.const.reg.apptype import AppType + + for proto in self._protos(): + with self.subTest(proto=proto.name): + field = self._field(proto) + self.assertEqual(field.pack(AppType.get(80, proto=proto), {}), b'\x00\x50') + self.assertEqual(field.pack(80, {}), b'\x00\x50') + + def test_unpack_then_pack_returns_the_octets(self) -> None: + for proto in self._protos(): + for octets in (b'\x00\x50', b'\x02\x00', b'\xd4\x31', b'\xff\xff', b'\x00\x00'): + with self.subTest(proto=proto.name, octets=octets): + field = self._field(proto) + self.assertEqual(field.pack(field.unpack(octets, {}), {}), octets) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/corekit/test_fields_numbers_port_option_no_mint_unit.py b/tests/corekit/test_fields_numbers_port_option_no_mint_unit.py index 4af7c6a34e..0099183046 100644 --- a/tests/corekit/test_fields_numbers_port_option_no_mint_unit.py +++ b/tests/corekit/test_fields_numbers_port_option_no_mint_unit.py @@ -2,7 +2,7 @@ """A registry-backed ``EnumField`` subclass minted a member per unassigned value. GitHub issue #575. :meth:`PortEnumField.post_process -` (TCP, UDP, +` (TCP, UDP, SCTP) and :meth:`OptionEnumField.post_process ` (PCAP-NG) each resolved a wire value by calling their registry's ``.get()`` directly:: @@ -93,14 +93,11 @@ def setUp(self) -> None: def _fields(self) -> 'list[tuple[str, Any, Any]]': from pcapkit.const.reg.apptype import AppType, TransportProtocol - from pcapkit.protocols.schema.transport import sctp as sctp_schema - from pcapkit.protocols.schema.transport import tcp as tcp_schema - from pcapkit.protocols.schema.transport import udp as udp_schema + from pcapkit.corekit.fields.numbers import PortEnumField return [ - ('tcp', tcp_schema.PortEnumField(length=2, namespace=AppType), TransportProtocol.tcp), - ('udp', udp_schema.PortEnumField(length=2, namespace=AppType), TransportProtocol.udp), - ('sctp', sctp_schema.PortEnumField(length=2, namespace=AppType), TransportProtocol.sctp), + (proto.name, PortEnumField(length=2, namespace=AppType, proto=proto), proto) + for proto in (TransportProtocol.tcp, TransportProtocol.udp, TransportProtocol.sctp) ] def test_an_ephemeral_port_resolves_without_growing_the_registry(self) -> None: @@ -269,7 +266,7 @@ def test_an_in_library_rejection_is_not_absorbed(self) -> None: """ from pcapkit.const.reg.apptype import AppType, TransportProtocol - from pcapkit.protocols.schema.transport import tcp as tcp_schema + from pcapkit.corekit.fields.numbers import PortEnumField from pcapkit.utilities.exceptions import FieldValueError owner = AppType.__registries__[TransportProtocol.tcp] @@ -281,7 +278,7 @@ def _reject(cls: object, value: int) -> None: owner._missing_ = classmethod(_reject) try: - field = tcp_schema.PortEnumField(length=2, namespace=AppType) + field = PortEnumField(length=2, namespace=AppType, proto=TransportProtocol.tcp) with self.assertRaises(FieldValueError): field.unpack(b'\xd4\x31', {}) finally: @@ -292,10 +289,10 @@ def _reject(cls: object, value: int) -> None: def test_the_resolved_pseudo_member_repacks_to_the_octets_it_came_from(self) -> None: """The fallback is round-trip safe, same as #701's.""" - from pcapkit.const.reg.apptype import AppType - from pcapkit.protocols.schema.transport import tcp as tcp_schema + from pcapkit.const.reg.apptype import AppType, TransportProtocol + from pcapkit.corekit.fields.numbers import PortEnumField - field = tcp_schema.PortEnumField(length=2, namespace=AppType) + field = PortEnumField(length=2, namespace=AppType, proto=TransportProtocol.tcp) resolved = field.unpack(b'\xd4\x31', {}) self.assertEqual(field.pack(resolved, {}), b'\xd4\x31') @@ -449,10 +446,10 @@ def setUp(self) -> None: def _resolved(self) -> 'Any': """An unassigned TCP port, resolved through the real field.""" - from pcapkit.const.reg.apptype import AppType - from pcapkit.protocols.schema.transport import tcp as tcp_schema + from pcapkit.const.reg.apptype import AppType, TransportProtocol + from pcapkit.corekit.fields.numbers import PortEnumField - field = tcp_schema.PortEnumField(length=2, namespace=AppType) + field = PortEnumField(length=2, namespace=AppType, proto=TransportProtocol.tcp) return field.unpack(b'\xd0\x9e', {}) # 53406 def test_a_pickle_round_trip_rebuilds_an_equivalent_unregistered_member(self) -> None: diff --git a/tests/dumpkit/test_common_unit.py b/tests/dumpkit/test_common_unit.py index 68a7eefdf2..a15d2f6829 100644 --- a/tests/dumpkit/test_common_unit.py +++ b/tests/dumpkit/test_common_unit.py @@ -359,11 +359,11 @@ def test_an_unassigned_port_dumps_its_addon_keys_in_the_declared_order(self) -> """ from pcapkit.const.reg.apptype import AppType, TransportProtocol + from pcapkit.corekit.fields.numbers import PortEnumField from pcapkit.dumpkit.common import make_dumper - from pcapkit.protocols.schema.transport import tcp as tcp_schema dumper = make_dumper(BaseDumper)() - field = tcp_schema.PortEnumField(length=2, namespace=AppType) + field = PortEnumField(length=2, namespace=AppType, proto=TransportProtocol.tcp) declared = dumper.object_hook(AppType.get(80, proto=TransportProtocol.tcp)) unassigned = dumper.object_hook(field.unpack(b'\xd4\x31', {})) # 54321 diff --git a/tests/protocols/transport/test_tcp_udp_unit.py b/tests/protocols/transport/test_tcp_udp_unit.py index cae9d97a34..a32ab1a089 100644 --- a/tests/protocols/transport/test_tcp_udp_unit.py +++ b/tests/protocols/transport/test_tcp_udp_unit.py @@ -1187,11 +1187,11 @@ def test_transport_schema_helpers_cover_selectors_and_port_fields(self) -> None: from pcapkit.const.reg.apptype import AppType, TransportProtocol from pcapkit.const.tcp.mp_tcp_option import MPTCPOption from pcapkit.const.tcp.option import Option + from pcapkit.corekit.fields.numbers import PortEnumField from pcapkit.protocols.schema.transport import tcp as tcp_schema - from pcapkit.protocols.schema.transport import udp as udp_schema from pcapkit.utilities.exceptions import FieldError - tcp_field = tcp_schema.PortEnumField(length=2, namespace=AppType) + tcp_field = PortEnumField(length=2, namespace=AppType, proto=TransportProtocol.tcp) tcp_field.name = 'srcport' self.assertEqual( tcp_field.pack(AppType.get(443, proto=TransportProtocol.tcp), {}), @@ -1200,7 +1200,7 @@ def test_transport_schema_helpers_cover_selectors_and_port_fields(self) -> None: self.assertEqual(tcp_field.pack(443, {}), b'\x01\xbb') self.assertEqual(tcp_field.unpack(b'\x01\xbb', {}).port, 443) - udp_field = udp_schema.PortEnumField(length=2, namespace=AppType) + udp_field = PortEnumField(length=2, namespace=AppType, proto=TransportProtocol.udp) udp_field.name = 'dstport' self.assertEqual( udp_field.pack(AppType.get(53, proto=TransportProtocol.udp), {}),