diff --git a/docs/source/pcapkit/protocols/protocol.rst b/docs/source/pcapkit/protocols/protocol.rst index 26de9066a2..16b95f3533 100644 --- a/docs/source/pcapkit/protocols/protocol.rst +++ b/docs/source/pcapkit/protocols/protocol.rst @@ -85,6 +85,23 @@ Internal Definitions below. .. automethod:: _make_payload .. automethod:: _packing_target + .. automethod:: _read_unpack + .. automethod:: _read_octets + .. automethod:: _read_uint + .. automethod:: _read_sint + .. automethod:: _read_vector + .. automethod:: _read_whole_vector + .. automethod:: _read_uints + + .. automethod:: _make_pack + .. automethod:: _make_octets + .. automethod:: _fill_short_read + .. automethod:: _make_uint + .. automethod:: _make_sint + .. automethod:: _uint_size + .. automethod:: _make_vector + .. automethod:: _make_uints + .. automethod:: _read_mac_addr .. automethod:: _make_mac_addr @@ -140,6 +157,29 @@ The ``pyshark``, ``scapy`` and ``pypcapfile`` engines dissect with their own parsers, which do not apply the frame's layer bound (:issue:`1609`). The ``dpkt`` toolkit reads what |dpkt|_ decodes, which applies neither bound. +Value Helpers +------------- + +A protocol reads and makes its field values through the helpers +:class:`~pcapkit.protocols.protocol.Protocol` carries, rather than through +copies of its own (:issue:`2614`). A maker checks a caller's values through them +before packing, and raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` for +one its field cannot carry rather than masking it (:issue:`2617`): + +* an unsigned integer through :meth:`~Protocol._make_uint`, which takes a + :obj:`bool` for a one-bit field only; +* a two's-complement integer through :meth:`~Protocol._make_sint`, the inverse of + :meth:`~Protocol._read_sint`; +* the octets of a fixed-size field through :meth:`~Protocol._make_octets`, which + refuses any other width rather than padding or cutting it (:issue:`2676`). A + value taken from a model of a cut capture goes through + :meth:`~Protocol._fill_short_read` first, which zero-fills just the octets the + cut left out, for the rebuild to cut back. + +:meth:`~Protocol._uint_size` counts the octets a variable-width integer needs, +and :func:`~pcapkit.corekit.fields.strings.pad_length` the padding that takes a +length to an alignment boundary. + Data Models ----------- @@ -205,6 +245,21 @@ Internal Definitions _make_index _make_payload _packing_target + _read_unpack + _read_octets + _read_uint + _read_sint + _read_vector + _read_whole_vector + _read_uints + _make_pack + _make_octets + _fill_short_read + _make_uint + _make_sint + _uint_size + _make_vector + _make_uints _read_mac_addr _make_mac_addr _lookup_registry @@ -233,7 +288,15 @@ Internal Definitions :meth:`~Protocol._read_packet`, :meth:`~Protocol._get_payload`, :meth:`~Protocol._make_data`, :meth:`~Protocol._make_uint`, :meth:`~Protocol._make_index`, :meth:`~Protocol._make_payload`, - :meth:`~Protocol._packing_target`, :meth:`~Protocol._read_mac_addr`, + :meth:`~Protocol._packing_target`, :meth:`~Protocol._read_unpack`, + :meth:`~Protocol._read_octets`, :meth:`~Protocol._read_uint`, + :meth:`~Protocol._read_sint`, :meth:`~Protocol._read_vector`, + :meth:`~Protocol._read_whole_vector`, :meth:`~Protocol._read_uints`, + :meth:`~Protocol._make_pack`, :meth:`~Protocol._make_octets`, + :meth:`~Protocol._fill_short_read`, + :meth:`~Protocol._make_uint`, :meth:`~Protocol._make_sint`, + :meth:`~Protocol._uint_size`, :meth:`~Protocol._make_vector`, + :meth:`~Protocol._make_uints`, :meth:`~Protocol._read_mac_addr`, :meth:`~Protocol._make_mac_addr`, :meth:`~Protocol._lookup_registry`, :meth:`~Protocol._lookup_next_layer`, :meth:`~Protocol._decode_next_layer`, diff --git a/pcapkit/protocols/application/ospf.py b/pcapkit/protocols/application/ospf.py index aab440d624..896526e67b 100644 --- a/pcapkit/protocols/application/ospf.py +++ b/pcapkit/protocols/application/ospf.py @@ -65,7 +65,13 @@ verified on read under :data:`~pcapkit.utilities.config.CHECKSUM_VERIFY` (:issue:`2475`). + :meth:`~pcapkit.protocols.application.ospf.OSPF.make` raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError` for a checksum, or the + reserved octets of cryptographic authentication data, that is not two + octets, rather than padding or cutting it (:issue:`2676`). + """ +import collections.abc import ipaddress from typing import TYPE_CHECKING, cast @@ -301,11 +307,12 @@ def make(self, Raises: ProtocolError: If ``auth_data`` is of an invalid type, is - :obj:`bytes` that are not exactly 8 octets long, or is a - :obj:`dict` missing a required key; if ``packet_length`` - is not an :obj:`int` in ``0..65535``; or if ``type`` (8 bits) - or ``auth_type`` (16 bits) is a name, a :obj:`bool`, or a - member or code its field cannot carry. + :obj:`bytes` that are not exactly 8 octets long, is a + :obj:`dict` missing a required key, or carries ``reserved`` + octets that are not two; if ``packet_length`` is not an + :obj:`int` in ``0..65535``; if ``checksum`` is not two octets; + or if ``type`` (8 bits) or ``auth_type`` (16 bits) is a name, a + :obj:`bool`, or a member or code its field cannot carry. .. versionchanged:: 1.5.0 @@ -341,7 +348,8 @@ def make(self, length=packet_length, router_id=router_id, area_id=area_id, - checksum=b'\x00\x00' if checksum is None else checksum, + checksum=(b'\x00\x00' if checksum is None + else self._make_octets(checksum, size=2, name='checksum', alias='OSPF')), auth_type=auth_type_, auth_data=data, payload=payload, @@ -391,6 +399,14 @@ def _make_data(cls, data: 'Data_OSPF') -> 'dict[str, Any]': # type: ignore[over Key-value pairs for protocol construction. """ + auth = data.auth # type: Any + # NOTE: Reserved octets a capture cut inside are zero-filled (#1458), in + # the data model or the dict form ``to_dict()`` gives of it. + if isinstance(auth, collections.abc.Mapping): + reserved = auth.get('reserved', b'\x00\x00') + filled = cls._fill_short_read(reserved, size=2, model=data, field='auth_data') + if filled is not reserved: + auth = {**auth, 'reserved': filled} return { 'version': data.version, 'type': data.type, @@ -399,7 +415,7 @@ def _make_data(cls, data: 'Data_OSPF') -> 'dict[str, Any]': # type: ignore[over 'area_id': data.area_id, 'checksum': data.chksum, 'auth_type': data.autype, - 'auth_data': data.auth, + 'auth_data': auth, 'payload': cls._make_payload(data), # NOTE: And the octets captured past the Packet Length (#1455). 'trailer': data.get('trailer', b''), @@ -497,8 +513,9 @@ def _make_encrypt_auth(self, Raises: ProtocolError: If ``auth_data`` is of an invalid type, is - :obj:`bytes` that are not exactly 8 octets long, or is a - :obj:`dict` missing a required key. + :obj:`bytes` that are not exactly 8 octets long, is a + :obj:`dict` missing a required key, or carries ``reserved`` + octets that are not two. """ if isinstance(auth_data, Schema_CryptographicAuthentication): @@ -509,7 +526,8 @@ def _make_encrypt_auth(self, return Schema_CryptographicAuthentication.unpack(auth_data) # type: ignore[call-arg,misc] if isinstance(auth_data, Data_CryptographicAuthentication): return Schema_CryptographicAuthentication( - reserved=getattr(auth_data, 'reserved', b'\x00\x00'), + reserved=self._make_octets(getattr(auth_data, 'reserved', b'\x00\x00'), size=2, + name='reserved octets', alias='OSPF'), key_id=auth_data.key_id, len=auth_data.len, seq=auth_data.seq, @@ -519,7 +537,8 @@ def _make_encrypt_auth(self, if isinstance(auth_data, dict): try: return Schema_CryptographicAuthentication( - reserved=auth_data.get('reserved', b'\x00\x00'), + reserved=self._make_octets(auth_data.get('reserved', b'\x00\x00'), size=2, + name='reserved octets', alias='OSPF'), key_id=auth_data['key_id'], len=auth_data['len'], seq=auth_data['seq'], diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index e7c8a44083..2224ed4ebb 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -452,6 +452,12 @@ class HIP(IPv6_Ext[Data_HIP, Schema_HIP], Internet[Data_HIP, Schema_HIP], :meth:`~pcapkit.protocols.protocol.Protocol.resolve_index` first (:issue:`2594`). + The same makers raise it for a fixed-size octet argument of the wrong + width -- the checksum, the ``PUZZLE`` and ``SOLUTION`` opaque data, the + ``PAYLOAD_MIC`` payload data, or reserved octets -- rather than padding + or cutting it to the field. The opaque and payload data default to + zeros of that width (:issue:`2676`). + """ ########################################################################## @@ -710,8 +716,8 @@ def make(self, reserved bit, ``shit`` or ``rhit`` does not fit its field, ``next`` or ``packet`` is a name, the code of a parameter in ``parameters`` is a name, a :obj:`bool` or a code its field - cannot carry, or the parameters make a header too long for its - one-octet Header Length. + cannot carry, ``checksum`` is not two octets, or the parameters + make a header too long for its one-octet Header Length. .. versionchanged:: 1.5.0 @@ -752,7 +758,7 @@ def make(self, 'reserved': self._make_uint(version_reserved, bits=3, name='version reserved bits', alias=f'HIPv{version}'), }, - checksum=checksum, + checksum=self._make_octets(checksum, size=2, name='checksum', alias=f'HIPv{version}'), control = { 'reserved': self._make_uint(controls_reserved, bits=15, name='controls reserved bits', alias=f'HIPv{version}'), @@ -3443,8 +3449,7 @@ def _make_param_r1_counter(self, code: 'Enum_Parameter', param: 'Optional[Data_R param: parameter data version: HIP protocol version counter: R1 generation counter - reserved: the four reserved octets, truncated or - zero-filled to 4 octets + reserved: the four reserved octets **kwargs: arbitrary keyword arguments Returns: @@ -3453,6 +3458,7 @@ def _make_param_r1_counter(self, code: 'Enum_Parameter', param: 'Optional[Data_R Raises: ProtocolError: If ``code`` is the HIPv1-only ``128`` and ``version`` is **NOT** ``1``, or ``counter`` does not fit in eight octets. + Also if ``reserved`` is not four octets. """ if code == Enum_Parameter.R1_Counter and version != 1: @@ -3463,7 +3469,7 @@ def _make_param_r1_counter(self, code: 'Enum_Parameter', param: 'Optional[Data_R counter = param.counter return Schema_R1CounterParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=4, name='reserved octets', alias=f'HIPv{version}'), type=code, len=12, counter=self._make_uint(counter, bits=64, name='R1 generation counter', @@ -3761,7 +3767,7 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl index: 'int' = 0, lifetime: 'timedelta | int' = 0, lifetime_exponent: 'Optional[int]' = None, - opaque: 'bytes' = b'', + opaque: 'bytes' = b'\x00\x00', random: 'int' = 0, rhash_len: 'Optional[int]' = None, **kwargs: 'Any') -> 'Schema_PuzzleParameter': @@ -3792,6 +3798,7 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl invalid; see ``_make_puzzle_lifetime`` and ``_make_puzzle_field_width``. Also if ``index`` does not fit one octet, or ``random`` is negative or a :obj:`bool`. + Also if ``opaque`` is not two octets. """ if param is not None: @@ -3818,7 +3825,7 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl len=4 + width, index=self._make_uint(index, bits=8, name='puzzle difficulty', alias=f'HIPv{version}'), lifetime=lifetime_exponent, - opaque=opaque, + opaque=self._make_octets(opaque, size=2, name='puzzle opaque', alias=f'HIPv{version}'), random=self._make_uint(random, bits=8 * width, name='puzzle random', alias=f'HIPv{version}'), ) @@ -3827,7 +3834,7 @@ def _make_param_solution(self, code: 'Enum_Parameter', param: 'Optional[Data_Sol version: 'int', index: 'int' = 0, reserved: 'Optional[int]' = None, - opaque: 'bytes' = b'', + opaque: 'bytes' = b'\x00\x00', random: 'int' = 0, solution: 'int' = 0, rhash_len: 'Optional[int]' = None, @@ -3863,7 +3870,7 @@ def _make_param_solution(self, code: 'Enum_Parameter', param: 'Optional[Data_Sol ProtocolError: If the width of the two fields is invalid; see ``_make_puzzle_field_width``. Also if ``index`` or ``reserved`` does not fit one octet, or ``random`` or ``solution`` is - negative or a :obj:`bool`. + negative or a :obj:`bool`. Also if ``opaque`` is not two octets. """ if param is not None: @@ -3900,7 +3907,7 @@ def _make_param_solution(self, code: 'Enum_Parameter', param: 'Optional[Data_Sol index=self._make_uint(index, bits=8, name='puzzle difficulty', alias=f'HIPv{version}'), reserved=self._make_uint(reserved, bits=8, name='reserved octet', alias=f'HIPv{version}'), - opaque=opaque, + opaque=self._make_octets(opaque, size=2, name='solution opaque', alias=f'HIPv{version}'), random=self._make_uint(random, bits=8 * width, name='puzzle random', alias=f'HIPv{version}'), solution=self._make_uint(solution, bits=8 * width, name='puzzle solution', @@ -4142,8 +4149,7 @@ def _make_param_nat_traversal_mode(self, code: 'Enum_Parameter', param: 'Optiona param: parameter data version: HIP protocol version modes: list of mode ID - reserved: the two reserved octets, truncated or - zero-filled to 2 octets + reserved: the two reserved octets **kwargs: arbitrary keyword arguments Returns: @@ -4151,7 +4157,7 @@ def _make_param_nat_traversal_mode(self, code: 'Enum_Parameter', param: 'Optiona Raises: ProtocolError: If an entry of ``modes`` is a name or does not fit - its field. + its field, or ``reserved`` is not two octets. """ if param is not None: @@ -4163,7 +4169,7 @@ def _make_param_nat_traversal_mode(self, code: 'Enum_Parameter', param: 'Optiona for mode in modes or []] return Schema_NATTraversalModeParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets', alias=f'HIPv{version}'), type=code, len=2 + 2 * len(mode_id), modes=mode_id, @@ -4215,8 +4221,7 @@ def _make_param_encrypted(self, code: 'Enum_Parameter', param: 'Optional[Data_En iv: initialization vector, given for the AES ciphers and only for them data: encrypted data - reserved: the four reserved octets, truncated or - zero-filled to 4 octets + reserved: the four reserved octets **kwargs: arbitrary keyword arguments Returns: @@ -4227,6 +4232,7 @@ def _make_param_encrypted(self, code: 'Enum_Parameter', param: 'Optional[Data_En the cipher is ``AES-128-CBC`` or ``AES-256-CBC`` and ``iv`` is missing or **NOT** 16 octets, or ``iv`` is given with any other cipher, which takes none. + Also if ``reserved`` is not four octets. """ if param is not None: @@ -4249,7 +4255,7 @@ def _make_param_encrypted(self, code: 'Enum_Parameter', param: 'Optional[Data_En raise ProtocolError(f'HIPv{version}: [ParamNo {code}] IV is not used by cipher {cipher_id!r}') schema = Schema_EncryptedParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=4, name='reserved octets', alias=f'HIPv{version}'), type=code, len=4 + len(iv or b'') + len(data), iv=iv, @@ -4496,15 +4502,15 @@ def _make_param_notification(self, code: 'Enum_Parameter', param: 'Optional[Data version: HIP protocol version msg_type: notification message type msg: notification message - reserved: the two reserved octets, truncated or - zero-filled to 2 octets + reserved: the two reserved octets **kwargs: arbitrary keyword arguments Returns: HIP parameter schema. Raises: - ProtocolError: If ``msg_type`` is a name or does not fit its field. + ProtocolError: If ``msg_type`` is a name or does not fit its field, + or ``reserved`` is not two octets. """ if param is not None: @@ -4516,7 +4522,7 @@ def _make_param_notification(self, code: 'Enum_Parameter', param: 'Optional[Data name='notify message type', alias=f'HIPv{version}') return Schema_NotificationParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets', alias=f'HIPv{version}'), type=code, len=4 + len(msg), msg_type=type, @@ -4783,8 +4789,7 @@ def _make_param_reg_from(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg port: port number protocol: transport protocol address: IPv6 address - reserved: the reserved octet, truncated or - zero-filled to 1 octet + reserved: the reserved octet **kwargs: arbitrary keyword arguments Returns: @@ -4793,6 +4798,7 @@ def _make_param_reg_from(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg Raises: ProtocolError: If ``port`` or ``protocol`` does not fit its field, or ``protocol`` is a name. + Also if ``reserved`` is not one octet. """ if param is not None: @@ -4805,7 +4811,7 @@ def _make_param_reg_from(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg alias=f'HIPv{version}') return Schema_RegFromParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets', alias=f'HIPv{version}'), type=code, len=20, port=self._make_uint(port, bits=16, name='port', alias=f'HIPv{version}'), @@ -4888,8 +4894,7 @@ def _make_param_esp_transform(self, code: 'Enum_Parameter', param: 'Optional[Dat param: parameter data version: HIP protocol version suites: ESP transform suite list - reserved: the two reserved octets, truncated or - zero-filled to 2 octets + reserved: the two reserved octets **kwargs: arbitrary keyword arguments Returns: @@ -4897,7 +4902,7 @@ def _make_param_esp_transform(self, code: 'Enum_Parameter', param: 'Optional[Dat Raises: ProtocolError: If an entry of ``suites`` is a name or does not fit - its field. + its field, or ``reserved`` is not two octets. """ if param is not None: @@ -4909,7 +4914,7 @@ def _make_param_esp_transform(self, code: 'Enum_Parameter', param: 'Optional[Dat for suite in suites or []] return Schema_ESPTransformParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets', alias=f'HIPv{version}'), type=code, len=2 + 2 * len(suite_id), suites=suite_id, @@ -4982,7 +4987,7 @@ def _make_param_ack_data(self, code: 'Enum_Parameter', param: 'Optional[Data_Ack def _make_param_payload_mic(self, code: 'Enum_Parameter', param: 'Optional[Data_PayloadMICParameter]' = None, *, # pylint: disable=unused-argument version: 'int', next: 'Enum_TransType | int' = Enum_TransType.UDP, - payload: 'bytes' = b'', + payload: 'bytes' = b'\x00\x00\x00\x00', mic: 'bytes' = b'', reserved: 'bytes' = b'\x00\x00\x00', **kwargs: 'Any') -> 'Schema_PayloadMICParameter': @@ -4995,15 +5000,15 @@ def _make_param_payload_mic(self, code: 'Enum_Parameter', param: 'Optional[Data_ next: next protocol payload: payload data mic: message integrity code - reserved: the three reserved octets, truncated or - zero-filled to 3 octets + reserved: the three reserved octets **kwargs: arbitrary keyword arguments Returns: HIP parameter schema. Raises: - ProtocolError: If ``next`` is a name or does not fit its field. + ProtocolError: If ``next`` is a name or does not fit its field, + ``payload`` is not four octets, or ``reserved`` is not three. """ if param is not None: @@ -5016,11 +5021,11 @@ def _make_param_payload_mic(self, code: 'Enum_Parameter', param: 'Optional[Data_ alias=f'HIPv{version}') return Schema_PayloadMICParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=3, name='reserved octets', alias=f'HIPv{version}'), type=code, len=8 + len(mic), next=protocol, - payload=payload, + payload=self._make_octets(payload, size=4, name='payload data', alias=f'HIPv{version}'), mic=mic, ) @@ -5177,8 +5182,7 @@ def _make_param_route_dst(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro must_follow: must-follow flag flags_reserved: the 14 reserved low-order bits of ``Flags`` hit: list of HITs - reserved: the two reserved octets, truncated or - zero-filled to 2 octets + reserved: the two reserved octets **kwargs: arbitrary keyword arguments Returns: @@ -5186,6 +5190,7 @@ def _make_param_route_dst(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro Raises: ProtocolError: If ``flags_reserved`` does not fit its field. + Also if ``reserved`` is not two octets. """ if param is not None: @@ -5198,7 +5203,7 @@ def _make_param_route_dst(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro hit_list = hit if hit is not None else [] return Schema_RouteDstParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets', alias=f'HIPv{version}'), type=code, len=4 + 16 * len(hit_list), flags={ @@ -5441,8 +5446,7 @@ def _make_param_relay_from(self, code: 'Enum_Parameter', param: 'Optional[Data_R protocol: transport protocol address: the address the relay server received the relayed packet from (:rfc:`5770`) - reserved: the reserved octet, truncated or - zero-filled to 1 octet + reserved: the reserved octet **kwargs: arbitrary keyword arguments Returns: @@ -5451,6 +5455,7 @@ def _make_param_relay_from(self, code: 'Enum_Parameter', param: 'Optional[Data_R Raises: ProtocolError: If ``port`` or ``protocol`` does not fit its field, or ``protocol`` is a name. + Also if ``reserved`` is not one octet. """ if param is not None: @@ -5463,7 +5468,7 @@ def _make_param_relay_from(self, code: 'Enum_Parameter', param: 'Optional[Data_R alias=f'HIPv{version}') return Schema_RelayFromParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets', alias=f'HIPv{version}'), type=code, len=20, port=self._make_uint(port, bits=16, name='port', alias=f'HIPv{version}'), @@ -5488,8 +5493,7 @@ def _make_param_relay_to(self, code: 'Enum_Parameter', param: 'Optional[Data_Rel protocol: transport protocol address: the address the relay server forwards the packet to (:rfc:`5770`) - reserved: the reserved octet, truncated or - zero-filled to 1 octet + reserved: the reserved octet **kwargs: arbitrary keyword arguments Returns: @@ -5498,6 +5502,7 @@ def _make_param_relay_to(self, code: 'Enum_Parameter', param: 'Optional[Data_Rel Raises: ProtocolError: If ``port`` or ``protocol`` does not fit its field, or ``protocol`` is a name. + Also if ``reserved`` is not one octet. """ if param is not None: @@ -5510,7 +5515,7 @@ def _make_param_relay_to(self, code: 'Enum_Parameter', param: 'Optional[Data_Rel alias=f'HIPv{version}') return Schema_RelayToParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets', alias=f'HIPv{version}'), type=code, len=20, port=self._make_uint(port, bits=16, name='port', alias=f'HIPv{version}'), @@ -5533,8 +5538,7 @@ def _make_param_overlay_ttl(self, code: 'Enum_Parameter', param: 'Optional[Data_ hops rather than a duration -- see :attr:`OverlayTTLParameter.ttl ` - reserved: the two reserved octets, truncated or - zero-filled to 2 octets + reserved: the two reserved octets **kwargs: arbitrary keyword arguments Returns: @@ -5542,6 +5546,7 @@ def _make_param_overlay_ttl(self, code: 'Enum_Parameter', param: 'Optional[Data_ Raises: ProtocolError: If ``ttl`` does not fit its field. + Also if ``reserved`` is not two octets. .. versionchanged:: 1.5.0 @@ -5556,7 +5561,7 @@ def _make_param_overlay_ttl(self, code: 'Enum_Parameter', param: 'Optional[Data_ ttl_val = ttl return Schema_OverlayTTLParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets', alias=f'HIPv{version}'), type=code, len=4, ttl=self._make_uint(ttl_val, bits=16, name='TTL', alias=f'HIPv{version}'), @@ -5580,8 +5585,7 @@ def _make_param_route_via(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro must_follow: must-follow flag flags_reserved: the 14 reserved low-order bits of ``Flags`` hit: list of HITs - reserved: the two reserved octets, truncated or - zero-filled to 2 octets + reserved: the two reserved octets **kwargs: arbitrary keyword arguments Returns: @@ -5589,6 +5593,7 @@ def _make_param_route_via(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro Raises: ProtocolError: If ``flags_reserved`` does not fit its field. + Also if ``reserved`` is not two octets. """ if param is not None: @@ -5601,7 +5606,7 @@ def _make_param_route_via(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro hit_list = hit if hit is not None else [] return Schema_RouteViaParameter( - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets', alias=f'HIPv{version}'), type=code, len=4 + 16 * len(hit_list), flags={ diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index c2d44e9326..9af668fbeb 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -230,6 +230,10 @@ class HOPOPT(IPv6_Ext[Data_HOPOPT, Schema_HOPOPT], resolving against an enumeration inferred from the caller's frame, writing a different value or raising another exception (:issue:`2594`). + * The CALIPSO option's checksum and the "Not Used" octet of a + Quick-Start report, given the wrong number of octets, raise + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than being + padded or cut to their field (:issue:`2676`). .. versionremoved:: 1.6.0 @@ -1683,7 +1687,8 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti Raises: ProtocolError: If ``domain`` or ``level`` does not fit its field, - if the bitmap is not bytes-like or an iterable of octets, or is + if ``checksum`` is not two octets, if the bitmap is not + bytes-like or an iterable of octets, or is not a whole number of 32-bit words, if ``pad`` is not :obj:`bytes`, or if the option data would exceed the 255 octets ``Opt Data Len`` can count. @@ -1733,7 +1738,7 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti domain=domain, cmpt_len=len(bitmap_bytes) // 4, level=level, - checksum=checksum, + checksum=self._make_octets(checksum, size=2, name='CALIPSO checksum'), bitmap=bitmap_bytes if bitmap_bytes else None, pad=pad, ) @@ -1939,8 +1944,8 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption Raises: ProtocolError: If ``func`` is a name, a :obj:`bool`, or a member or code that does not fit in four bits, the rate code cannot carry - ``rate``, or ``ttl``, ``nonce`` or ``reserved`` does not fit its - field. + ``rate``, ``ttl``, ``nonce`` or ``reserved`` does not fit its + field, or ``unused`` is not one octet. """ if isinstance(opt, Data_UnassignedOption): @@ -1981,7 +1986,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption 'func': func_enum, 'rate': rate_val, }, - reserved=unused, + reserved=self._make_octets(unused, size=1, name='Quick-Start not-used octet'), nonce={ 'nonce': nonce, 'reserved': reserved, diff --git a/pcapkit/protocols/internet/icmpv4.py b/pcapkit/protocols/internet/icmpv4.py index 0e8978a26d..dc693f3ad4 100644 --- a/pcapkit/protocols/internet/icmpv4.py +++ b/pcapkit/protocols/internet/icmpv4.py @@ -241,9 +241,10 @@ def make(self, Raises: ProtocolError: If ``type`` is a name, a :obj:`bool`, or a member - or code that does not fit in one octet; or if ``body`` is + or code that does not fit in one octet; if ``body`` is :obj:`bytes` that do not parse as the message body of - ``type``, or is of none of the accepted types. + ``type``, or is of none of the accepted types; or if + ``checksum`` is not two octets. """ type_val = self._make_index(type, Enum_Type, bits=8, name='type') @@ -276,7 +277,7 @@ def make(self, schema = Schema_ICMPv4( type=type_val, code=code, - checksum=b'\x00\x00' if checksum is None else checksum, + checksum=b'\x00\x00' if checksum is None else self._make_octets(checksum, size=2, name='checksum'), data=data_val, ) @@ -562,8 +563,7 @@ def _make_data_type_destination_unreachable(self, type: 'Enum_Type', # pylint: Args: type: ICMP type. message: message data. - unused: the four unused octets, truncated or zero-filled to four - octets. + unused: the four unused octets. data: internet header plus the first eight octets of the original datagram's payload. **kwargs: arbitrary keyword arguments. @@ -571,11 +571,15 @@ def _make_data_type_destination_unreachable(self, type: 'Enum_Type', # pylint: Returns: Constructed message body schema. + Raises: + ProtocolError: If ``unused`` is not four octets. + """ if message is not None: unused = getattr(message, 'unused', b'\x00\x00\x00\x00') data = message.data - return Schema_DestinationUnreachable(unused=unused, data=data) + return Schema_DestinationUnreachable(unused=self._make_octets(unused, size=4, name='unused octets'), + data=data) def _make_data_type_time_exceeded(self, type: 'Enum_Type', # pylint: disable=redefined-builtin,unused-argument message: 'Optional[Data_TimeExceeded]' = None, *, @@ -586,8 +590,7 @@ def _make_data_type_time_exceeded(self, type: 'Enum_Type', # pylint: disable=re Args: type: ICMP type. message: message data. - unused: the four unused octets, truncated or zero-filled to four - octets. + unused: the four unused octets. data: internet header plus the first eight octets of the original datagram's payload. **kwargs: arbitrary keyword arguments. @@ -595,11 +598,15 @@ def _make_data_type_time_exceeded(self, type: 'Enum_Type', # pylint: disable=re Returns: Constructed message body schema. + Raises: + ProtocolError: If ``unused`` is not four octets. + """ if message is not None: unused = getattr(message, 'unused', b'\x00\x00\x00\x00') data = message.data - return Schema_TimeExceeded(unused=unused, data=data) + return Schema_TimeExceeded(unused=self._make_octets(unused, size=4, name='unused octets'), + data=data) def _make_data_type_parameter_problem(self, type: 'Enum_Type', # pylint: disable=redefined-builtin,unused-argument message: 'Optional[Data_ParameterProblem]' = None, *, @@ -612,8 +619,7 @@ def _make_data_type_parameter_problem(self, type: 'Enum_Type', # pylint: disabl message: message data. pointer: pointer to the octet of the original datagram that caused the error. - unused: the three unused octets, truncated or zero-filled to - three octets. + unused: the three unused octets. data: internet header plus the first eight octets of the original datagram's payload. **kwargs: arbitrary keyword arguments. @@ -621,9 +627,14 @@ def _make_data_type_parameter_problem(self, type: 'Enum_Type', # pylint: disabl Returns: Constructed message body schema. + Raises: + ProtocolError: If ``unused`` is not three octets. + """ if message is not None: pointer = message.pointer unused = getattr(message, 'unused', b'\x00\x00\x00') data = message.data - return Schema_ParameterProblem(pointer=pointer, unused=unused, data=data) + return Schema_ParameterProblem(pointer=pointer, + unused=self._make_octets(unused, size=3, name='unused octets'), + data=data) diff --git a/pcapkit/protocols/internet/icmpv6.py b/pcapkit/protocols/internet/icmpv6.py index c1eeae2b4b..97485d4c00 100644 --- a/pcapkit/protocols/internet/icmpv6.py +++ b/pcapkit/protocols/internet/icmpv6.py @@ -246,9 +246,10 @@ def make(self, Raises: ProtocolError: If ``type`` is a name, a :obj:`bool`, or a member - or code that does not fit in one octet; or if ``body`` is + or code that does not fit in one octet; if ``body`` is :obj:`bytes` that do not parse as the message body of - ``type``, or is of none of the accepted types. + ``type``, or is of none of the accepted types; or if + ``checksum`` is not two octets. """ type_val = self._make_index(type, Enum_Type, bits=8, name='type') @@ -281,7 +282,7 @@ def make(self, schema = Schema_ICMPv6( type=type_val, code=code, - checksum=b'\x00\x00' if checksum is None else checksum, + checksum=b'\x00\x00' if checksum is None else self._make_octets(checksum, size=2, name='checksum'), data=data_val, ) @@ -608,8 +609,7 @@ def _make_data_type_destination_unreachable(self, type: 'Enum_Type', # pylint: Args: type: ICMPv6 type. message: message data. - unused: the four unused octets, truncated or zero-filled to four - octets. + unused: the four unused octets. data: as much of the invoking packet as fits the minimum IPv6 MTU. **kwargs: arbitrary keyword arguments. @@ -617,11 +617,15 @@ def _make_data_type_destination_unreachable(self, type: 'Enum_Type', # pylint: Returns: Constructed message body schema. + Raises: + ProtocolError: If ``unused`` is not four octets. + """ if message is not None: unused = getattr(message, 'unused', b'\x00\x00\x00\x00') data = message.data - return Schema_DestinationUnreachable(unused=unused, data=data) + return Schema_DestinationUnreachable(unused=self._make_octets(unused, size=4, name='unused octets'), + data=data) def _make_data_type_packet_too_big(self, type: 'Enum_Type', # pylint: disable=redefined-builtin,unused-argument message: 'Optional[Data_PacketTooBig]' = None, *, @@ -655,8 +659,7 @@ def _make_data_type_time_exceeded(self, type: 'Enum_Type', # pylint: disable=re Args: type: ICMPv6 type. message: message data. - unused: the four unused octets, truncated or zero-filled to four - octets. + unused: the four unused octets. data: as much of the invoking packet as fits the minimum IPv6 MTU. **kwargs: arbitrary keyword arguments. @@ -664,11 +667,15 @@ def _make_data_type_time_exceeded(self, type: 'Enum_Type', # pylint: disable=re Returns: Constructed message body schema. + Raises: + ProtocolError: If ``unused`` is not four octets. + """ if message is not None: unused = getattr(message, 'unused', b'\x00\x00\x00\x00') data = message.data - return Schema_TimeExceeded(unused=unused, data=data) + return Schema_TimeExceeded(unused=self._make_octets(unused, size=4, name='unused octets'), + data=data) def _make_data_type_parameter_problem(self, type: 'Enum_Type', # pylint: disable=redefined-builtin,unused-argument message: 'Optional[Data_ParameterProblem]' = None, *, diff --git a/pcapkit/protocols/internet/igmp.py b/pcapkit/protocols/internet/igmp.py index d97bc05159..9e774490f1 100644 --- a/pcapkit/protocols/internet/igmp.py +++ b/pcapkit/protocols/internet/igmp.py @@ -187,8 +187,9 @@ def make(self, Raises: ProtocolError: If ``type`` is a name, a :obj:`bool`, or a member - or code outside 0 to 255, or ``max_resp_time`` is not a - non-:obj:`bool` integer from 0 to 255. + or code outside 0 to 255, ``max_resp_time`` is not a + non-:obj:`bool` integer from 0 to 255, or ``checksum`` is not + two octets. """ type_value = self._make_index(type, Enum_Type, bits=8, name='type') @@ -197,7 +198,7 @@ def make(self, schema = Schema_IGMP( type=type_value, max_resp_time=max_resp_time, - checksum=b'\x00\x00' if checksum is None else checksum, + checksum=b'\x00\x00' if checksum is None else self._make_octets(checksum, size=2, name='checksum'), group=group, payload=payload, ) diff --git a/pcapkit/protocols/internet/igmpv3.py b/pcapkit/protocols/internet/igmpv3.py index ecd1de1256..54ebceedeb 100644 --- a/pcapkit/protocols/internet/igmpv3.py +++ b/pcapkit/protocols/internet/igmpv3.py @@ -388,8 +388,9 @@ def make(self, or code outside 0 to 255, or is neither a Query nor a Report; if ``max_resp_time`` is not an :obj:`int` from 0 to 255; if ``num_group_records`` is not an :obj:`int` from the length of - ``records`` to 65535; or if a record's ``record_type`` is - refused, see :meth:`_make_group_record`. + ``records`` to 65535; if ``checksum`` or ``reserved_2`` is not + two octets; or if a record's ``record_type`` is refused, see + :meth:`_make_group_record`. """ type_value = self._make_index(type, Enum_Type, bits=8, name='type') @@ -397,7 +398,7 @@ def make(self, or not 0 <= max_resp_time <= 255: raise ProtocolError(f'{self.alias}: invalid max_resp_time: {max_resp_time!r}') - field = b'\x00\x00' if checksum is None else checksum + field = b'\x00\x00' if checksum is None else self._make_octets(checksum, size=2, name='checksum') if type_value == Enum_Type.IGMPv3_Membership_Report: record_list = [] if records is None else [ self._make_group_record(record) if not isinstance(record, dict) else @@ -414,7 +415,7 @@ def make(self, body=Schema_Report( reserved_1=reserved_1, checksum=field, - reserved_2=reserved_2, + reserved_2=self._make_octets(reserved_2, size=2, name='reserved_2'), num_group_records=num_group_records, records=record_list, payload=payload, diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index 2573c6e3b3..9d6d1233aa 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -238,6 +238,11 @@ class IPv4(IP[Data_IPv4, Schema_IPv4], writing a different value or raising a bare :exc:`~pcapkit.utilities.exceptions.FieldValueError` (:issue:`2594`). + * The timestamp option's maker raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError` for a slot count + the option has no room for, rather than cutting it to nine or four, + and :meth:`make` for a checksum that is not two octets, rather than + padding or cutting it (:issue:`2676`). .. versionremoved:: 1.6.0 @@ -529,11 +534,11 @@ def make(self, Raises: ProtocolError: If ``id``, ``reserved``, ``offset``, ``ttl`` or the - total length does not fit its field; if a ToS field, - ``protocol`` or the type of an option is a name, a :obj:`bool` - for a field wider than one bit, or a member or code its field - cannot carry; or if the options make a header too long for its - four-bit IHL. + total length does not fit its field; if ``checksum`` is not + two octets; if a ToS field, ``protocol`` or the type of an + option is a name, a :obj:`bool` for a field wider than one bit, + or a member or code its field cannot carry; or if the options + make a header too long for its four-bit IHL. .. versionchanged:: 1.5.0 @@ -593,7 +598,7 @@ def make(self, }, ttl=ttl_val, proto=proto, - chksum=b'\x00\x00' if checksum is None else checksum, + chksum=b'\x00\x00' if checksum is None else self._make_octets(checksum, size=2, name='checksum'), src=src, dst=dst, options=options_value, @@ -1801,7 +1806,7 @@ def _make_opt_lsr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_LSROpt ) def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOption]' = None, *, - counts: 'int' = 5, + counts: 'Optional[int]' = None, overflow: 'int' = 0, timestamp: 'Optional[list[int | timedelta] | dict[IPv4Address | str | int, int | timedelta]]' = None, remaining: 'Optional[list[int]]' = None, @@ -1811,7 +1816,10 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio Args: kind: option type code option: option data - counts: maximum number of timestamps to record + counts: maximum number of timestamps to record, at most nine for + a :obj:`list` of timestamps and four for a :obj:`dict`, which + fill the option's 40 octets (:rfc:`791#section-3.1`); + :data:`None` is five, or four for a :obj:`dict` overflow: overflow count, i.e. the number of modules that could not register a timestamp (:rfc:`791#section-3.1`) timestamp: list of timestamps, i.e. the slots before the pointer, or @@ -1828,9 +1836,9 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio Raises: ProtocolError: If ``option`` is not given and ``timestamp`` is neither :data:`None`, a :obj:`list`, nor a :obj:`dict`, or - ``counts`` is not a non-negative :obj:`int`; or if ``overflow`` - does not fit in four bits, or a timestamp or a word of - ``remaining`` in 32 bits. + ``counts`` is not a non-negative :obj:`int` the option has + room for; or if ``overflow`` does not fit in four bits, or a + timestamp or a word of ``remaining`` in 32 bits. FieldValueError: If an address is a :obj:`bool` or is not a valid IPv4 address. @@ -1867,15 +1875,17 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio else: ts_list = [] remainder = b'' - # NOTE: Not _make_uint: a count past what the header holds is - # clamped to it below rather than refused, so it has no width. - if isinstance(counts, bool) or not isinstance(counts, int) or counts < 0: - raise ProtocolError(f'{self.alias}: invalid timestamp count: {counts!r}') if timestamp is None: timestamp = [] + # NOTE: The option is 40 octets at most (RFC 791): nine timestamps or + # four address pairs. Not _make_uint, as that bound is no field width. + room = 4 if isinstance(timestamp, dict) else 9 + if counts is None: + counts = min(5, room) + if isinstance(counts, bool) or not isinstance(counts, int) or not 0 <= counts <= room: + raise ProtocolError(f'{self.alias}: invalid timestamp count: {counts!r}') if isinstance(timestamp, list): flag = Enum_TSFlag.Timestamp_Only # type: ignore[assignment] - counts = min(9, counts) # 9 is the maximum number of timestamps length = 4 + counts * 4 for index, ts in enumerate(timestamp): @@ -1891,7 +1901,6 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio ts_list.append(ts) elif isinstance(timestamp, dict): flag = Enum_TSFlag.IP_with_Timestamp # type: ignore[assignment] - counts = min(4, counts) # 4 is the maximum number of timestamps length = 4 + counts * 8 for index, (addr, ts) in enumerate(timestamp.items()): diff --git a/pcapkit/protocols/internet/ipv6_frag.py b/pcapkit/protocols/internet/ipv6_frag.py index 51729b3302..2fbcb047f2 100644 --- a/pcapkit/protocols/internet/ipv6_frag.py +++ b/pcapkit/protocols/internet/ipv6_frag.py @@ -62,6 +62,10 @@ class IPv6_Frag(IPv6_Ext[Data_IPv6_Frag, Schema_IPv6_Frag], resolving against an enumeration inferred from the caller's frame or writing a different value (:issue:`2594`). + :meth:`make` raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` + for a ``reserved_octet`` that is not one octet, rather than padding or + cutting it (:issue:`2676`). + .. versionremoved:: 1.6.0 The ``next_default``, ``next_namespace`` and ``next_reversed`` @@ -192,8 +196,7 @@ def make(self, Args: next: Next header. - reserved_octet: The reserved octet after the next header, truncated or - zero-filled to one octet. + reserved_octet: The reserved octet after the next header. offset: Fragment offset, in on-wire 8-octet units (:rfc:`8200#section-4.5`). Note that the data model's :attr:`offset ` is in @@ -209,7 +212,8 @@ def make(self, Raises: ProtocolError: If ``next`` is a name, a :obj:`bool`, or a member - or code that does not fit in one octet. + or code that does not fit in one octet, or ``reserved_octet`` + is not one octet. .. versionchanged:: 1.5.0 @@ -222,7 +226,7 @@ def make(self, return Schema_IPv6_Frag( next=next_val, - reserved=reserved_octet, + reserved=self._make_octets(reserved_octet, size=1, name='reserved octet'), flags={ 'offset': offset, 'reserved': reserved, @@ -304,12 +308,15 @@ def _make_data(cls, data: 'Data_IPv6_Frag') -> 'dict[str, Any]': # type: ignore Key-value pairs for protocol construction. """ + # NOTE: A capture cut before the octet read none of it (#1458). + reserved_octet = cls._fill_short_read(getattr(data, 'reserved_octet', b'\x00'), + size=1, model=data, field='reserved') # NOTE: ``make`` takes ``offset`` in on-wire 8-octet units, while # ``data.offset`` is in octets (see :meth:`read`), so scale back down to # keep the data-to-schema round trip exact. return { 'next': data.next, - 'reserved_octet': getattr(data, 'reserved_octet', b'\x00'), + 'reserved_octet': reserved_octet, 'offset': data.offset // 8, 'reserved': data.reserved, 'mf': data.mf, diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index 037a93bb0a..f3cfa7d0ec 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -234,6 +234,10 @@ class IPv6_Opts(IPv6_Ext[Data_IPv6_Opts, Schema_IPv6_Opts], resolving against an enumeration inferred from the caller's frame, writing a different value or raising another exception (:issue:`2594`). + * The CALIPSO option's checksum and the "Not Used" octet of a + Quick-Start report, given the wrong number of octets, raise + :exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than being + padded or cut to their field (:issue:`2676`). .. versionremoved:: 1.6.0 @@ -1671,7 +1675,8 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti Raises: ProtocolError: If ``domain`` or ``level`` does not fit its field, - if the bitmap is not bytes-like or an iterable of octets, or is + if ``checksum`` is not two octets, if the bitmap is not + bytes-like or an iterable of octets, or is not a whole number of 32-bit words, if ``pad`` is not :obj:`bytes`, or if the option data would exceed the 255 octets ``Opt Data Len`` can count. @@ -1721,7 +1726,7 @@ def _make_opt_calipso(self, code: 'Enum_Option', opt: 'Optional[Data_CALIPSOOpti domain=domain, cmpt_len=len(bitmap_bytes) // 4, level=level, - checksum=checksum, + checksum=self._make_octets(checksum, size=2, name='CALIPSO checksum'), bitmap=bitmap_bytes if bitmap_bytes else None, pad=pad, ) @@ -1927,8 +1932,8 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption Raises: ProtocolError: If ``func`` is a name, a :obj:`bool`, or a member or code that does not fit in four bits, the rate code cannot carry - ``rate``, or ``ttl``, ``nonce`` or ``reserved`` does not fit its - field. + ``rate``, ``ttl``, ``nonce`` or ``reserved`` does not fit its + field, or ``unused`` is not one octet. """ if isinstance(opt, Data_UnassignedOption): @@ -1969,7 +1974,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption 'func': func_enum, 'rate': rate_val, }, - reserved=unused, + reserved=self._make_octets(unused, size=1, name='Quick-Start not-used octet'), nonce={ 'nonce': nonce, 'reserved': reserved, diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index e6e7ad7424..2ec85a3824 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/pcapkit/protocols/internet/ipv6_route.py @@ -138,6 +138,9 @@ class IPv6_Route(IPv6_Ext[Data_IPv6_Route, Schema_IPv6_Route], :meth:`~pcapkit.protocols.protocol.Protocol.resolve_index` first (:issue:`2594`). + The Source Route and Type 2 makers raise it for ``reserved`` octets that + are not four, rather than padding or cutting them (:issue:`2676`). + """ ########################################################################## @@ -790,22 +793,25 @@ def _make_data_type_src(self, type: 'Enum_Routing', route: 'Optional[Data_Source route: route data dst: destination IPv6 address ip: list of IPv6 addresses - reserved: the four reserved octets, truncated or zero-filled to - four octets + reserved: the four reserved octets **kwargs: arbitrary keyword arguments Returns: Constructed route data schema. + Raises: + ProtocolError: If ``reserved`` is not four octets. + """ if route is not None: ip = cast('list[IPv6Address | str | bytes | int]', route.ip) - reserved = getattr(route, 'reserved', b'\x00\x00\x00\x00') + reserved = self._fill_short_read(getattr(route, 'reserved', b'\x00\x00\x00\x00'), + size=4, model=route, field='data') else: ip = [] if ip is None else ip return Schema_SourceRoute( - reserved=reserved, + reserved=self._make_octets(reserved, size=4, name='reserved octets'), ip=ip, ) @@ -821,20 +827,23 @@ def _make_data_type_2(self, type: 'Enum_Routing', route: 'Optional[Data_Type2]' route: route data dst: destination IPv6 address ip: home address - reserved: the four reserved octets, truncated or zero-filled to - four octets + reserved: the four reserved octets **kwargs: arbitrary keyword arguments Returns: Constructed route data schema. + Raises: + ProtocolError: If ``reserved`` is not four octets. + """ if route is not None: ip = cast('IPv6Address | str | bytes | int', route.ip) - reserved = getattr(route, 'reserved', b'\x00\x00\x00\x00') + reserved = self._fill_short_read(getattr(route, 'reserved', b'\x00\x00\x00\x00'), + size=4, model=route, field='data') return Schema_Type2( - reserved=reserved, + reserved=self._make_octets(reserved, size=4, name='reserved octets'), ip=ip, ) diff --git a/pcapkit/protocols/internet/ipx.py b/pcapkit/protocols/internet/ipx.py index a01087a0f0..4cee4c8571 100644 --- a/pcapkit/protocols/internet/ipx.py +++ b/pcapkit/protocols/internet/ipx.py @@ -76,6 +76,10 @@ class IPX(Internet[Data_IPX, Schema_IPX], resolving against an enumeration inferred from the caller's frame or writing a different value (:issue:`2594`). + :meth:`make` raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` + for a checksum that is not two octets, or an address that is not + twelve, rather than padding or cutting it (:issue:`2676`). + .. versionremoved:: 1.6.0 The ``type_default``, ``type_namespace`` and ``type_reversed`` @@ -194,13 +198,13 @@ def make(self, """Make (construct) packet data. Args: - chksum: Checksum. + chksum: Checksum, two octets. count: Transport Control (hop count). type: Packet Type. total_length: Packet length (header included); computed as ``30`` plus the payload length when not given. - dst: Destination Address. - src: Source Address. + dst: Destination Address, twelve octets. + src: Source Address, twelve octets. payload: Payload data. trailer: Octets after the packet, outside the Packet Length, such as link-layer padding; written as is. @@ -211,7 +215,8 @@ def make(self, Raises: ProtocolError: If ``type`` is a name, a :obj:`bool`, or a member - or code that does not fit in one octet. + or code that does not fit in one octet, ``chksum`` is not two + octets, or ``dst`` or ``src`` is not twelve. .. versionchanged:: 1.5.0 @@ -224,12 +229,12 @@ def make(self, total_length = 30 + len(payload) return Schema_IPX( - chksum=chksum, + chksum=self._make_octets(chksum, size=2, name='checksum'), len=total_length, count=count, type=type_val, - dst=dst, - src=src, + dst=self._make_octets(dst, size=12, name='destination address'), + src=self._make_octets(src, size=12, name='source address'), payload=payload, trailer=trailer, ) diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index bfa7920c11..87c1fa4e10 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -1161,6 +1161,11 @@ class MH(IPv6_Ext[Data_MH, Schema_MH], its schema holds, rather than with :class:`~pcapkit.const.mh.status_code.StatusCode` (:issue:`2594`). + The same makers raise it for a fixed-size octet argument of the wrong + width -- the checksum, a cookie or keygen token, or reserved octets -- + rather than padding or cutting it to the field. The checksum defaults + to two zero octets (:issue:`2676`). + .. versionremoved:: 1.6.0 * ``_decode_signed``, the two's-complement decoder of the geo-location @@ -1451,7 +1456,7 @@ def make(self, next: 'Enum_TransType | int' = Enum_TransType.UDP, type: 'Enum_Packet | int' = Enum_Packet.Binding_Refresh_Request, reserved: 'int' = 0, - chksum: 'bytes' = b'', + chksum: 'bytes' = b'\x00\x00', data: 'bytes | Data_MH | Schema_Packet | dict[str, Any]' = b'\x00\x00', # minimum length payload: 'ProtocolBase | Schema | bytes' = b'', **kwargs: 'Any') -> 'Schema_MH': @@ -1461,7 +1466,7 @@ def make(self, next: Next header type. type: Mobility Header type. reserved: Reserved octet after the MH type. - chksum: Checksum. + chksum: Checksum, two octets. data: Message data. payload: Payload of next layer protocol. **kwargs: Arbitrary keyword arguments. @@ -1475,8 +1480,9 @@ def make(self, :class:`~pcapkit.protocols.data.internet.mh.MH` data model or a :class:`~pcapkit.protocols.schema.internet.mh.Packet` schema; or if ``reserved``, or the header length of the message, does - not fit in one octet; or if ``next`` or ``type`` is a name, a - :obj:`bool` or a code its field cannot carry. + not fit in one octet; if ``chksum`` is not two octets; or if + ``next`` or ``type`` is a name, a :obj:`bool` or a code its + field cannot carry. .. versionchanged:: 1.5.0 @@ -1526,7 +1532,7 @@ def make(self, length=self._make_uint(header_len, bits=8, name='header length'), type=type_val, reserved=self._make_uint(reserved, bits=8, name='reserved octet'), - chksum=chksum, + chksum=self._make_octets(chksum, size=2, name='checksum'), data=data_val, payload=payload, ) @@ -6830,24 +6836,28 @@ def _make_msg_hoti(self, message: 'Optional[Data_HomeTestInitMessage]' = None, * message: Message data model. cookie: Home test cookie. options: Mobility options. - msg_reserved: The two reserved octets of the message body as captured, - truncated or zero-filled to 2 octets. + msg_reserved: The two reserved octets of the message body as captured. **kwargs: Arbitrary keyword arguments. Returns: Constructed message type. + Raises: + ProtocolError: If ``cookie`` is not eight octets, or ``msg_reserved`` + is not two. + """ if message is not None: - msg_reserved = getattr(message, 'msg_reserved', b'\x00\x00') + msg_reserved = self._fill_short_read(getattr(message, 'msg_reserved', b'\x00\x00'), + size=2, model=message, field='data') cookie = message.cookie options = message.options else: options = options or [] return Schema_HomeTestInitMessage( - reserved=msg_reserved, - cookie=cookie, + reserved=self._make_octets(msg_reserved, size=2, name='reserved octets'), + cookie=self._make_octets(cookie, size=8, name='cookie'), options=self._make_mh_options(options), ) @@ -6862,24 +6872,28 @@ def _make_msg_coti(self, message: 'Optional[Data_CareofTestInitMessage]' = None, message: Message data model. cookie: Care-of test cookie. options: Mobility options. - msg_reserved: The two reserved octets of the message body as captured, - truncated or zero-filled to 2 octets. + msg_reserved: The two reserved octets of the message body as captured. **kwargs: Arbitrary keyword arguments. Returns: Constructed message type. + Raises: + ProtocolError: If ``cookie`` is not eight octets, or ``msg_reserved`` + is not two. + """ if message is not None: - msg_reserved = getattr(message, 'msg_reserved', b'\x00\x00') + msg_reserved = self._fill_short_read(getattr(message, 'msg_reserved', b'\x00\x00'), + size=2, model=message, field='data') cookie = message.cookie options = message.options else: options = options or [] return Schema_CareofTestInitMessage( - reserved=msg_reserved, - cookie=cookie, + reserved=self._make_octets(msg_reserved, size=2, name='reserved octets'), + cookie=self._make_octets(cookie, size=8, name='cookie'), options=self._make_mh_options(options), ) @@ -6904,6 +6918,7 @@ def _make_msg_hot(self, message: 'Optional[Data_HomeTestMessage]' = None, *, Raises: ProtocolError: If ``nonce_index`` does not fit its field. + Also if ``cookie`` or ``token`` is not eight octets. """ if message is not None: @@ -6916,8 +6931,8 @@ def _make_msg_hot(self, message: 'Optional[Data_HomeTestMessage]' = None, *, return Schema_HomeTestMessage( nonce_index=self._make_uint(nonce_index, bits=16, name='nonce index'), - cookie=cookie, - token=token, + cookie=self._make_octets(cookie, size=8, name='cookie'), + token=self._make_octets(token, size=8, name='token'), options=self._make_mh_options(options), ) @@ -6942,6 +6957,7 @@ def _make_msg_cot(self, message: 'Optional[Data_CareofTestMessage]' = None, *, Raises: ProtocolError: If ``nonce_index`` does not fit its field. + Also if ``cookie`` or ``token`` is not eight octets. """ if message is not None: @@ -6954,8 +6970,8 @@ def _make_msg_cot(self, message: 'Optional[Data_CareofTestMessage]' = None, *, return Schema_CareofTestMessage( nonce_index=self._make_uint(nonce_index, bits=16, name='nonce index'), - cookie=cookie, - token=token, + cookie=self._make_octets(cookie, size=8, name='cookie'), + token=self._make_octets(token, size=8, name='token'), options=self._make_mh_options(options), ) @@ -7151,8 +7167,7 @@ def _make_msg_be(self, message: 'Optional[Data_BindingErrorMessage]' = None, *, code no status is assigned. home: Home address. options: Mobility options. - msg_reserved: The reserved octet of the message body as captured, - truncated or zero-filled to 1 octet. + msg_reserved: The reserved octet of the message body as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -7160,11 +7175,12 @@ def _make_msg_be(self, message: 'Optional[Data_BindingErrorMessage]' = None, *, Raises: ProtocolError: If ``status`` is a name, a :obj:`bool` or a code its - field cannot carry. + field cannot carry, or ``msg_reserved`` is not one octet. """ if message is not None: - msg_reserved = getattr(message, 'msg_reserved', b'\x00') + msg_reserved = self._fill_short_read(getattr(message, 'msg_reserved', b'\x00'), + size=1, model=message, field='data', offset=1) status_val = message.status home = message.home options = message.options @@ -7174,7 +7190,7 @@ def _make_msg_be(self, message: 'Optional[Data_BindingErrorMessage]' = None, *, return Schema_BindingErrorMessage( status=status_val, - reserved=msg_reserved, + reserved=self._make_octets(msg_reserved, size=1, name='reserved octets'), home=home, options=self._make_mh_options(options), ) @@ -7306,22 +7322,25 @@ def _make_msg_fna(self, message: 'Optional[Data_FastNeighborAdvertisementMessage Args: message: Message data model. options: Mobility options. - msg_reserved: The two reserved octets of the message body as captured, - truncated or zero-filled to 2 octets. + msg_reserved: The two reserved octets of the message body as captured. **kwargs: Arbitrary keyword arguments. Returns: Constructed message type. + Raises: + ProtocolError: If ``msg_reserved`` is not two octets. + """ if message is not None: - msg_reserved = getattr(message, 'msg_reserved', b'\x00\x00') + msg_reserved = self._fill_short_read(getattr(message, 'msg_reserved', b'\x00\x00'), + size=2, model=message, field='data') options = message.options else: options = options or [] return Schema_FastNeighborAdvertisementMessage( - reserved=msg_reserved, + reserved=self._make_octets(msg_reserved, size=2, name='reserved octets'), options=self._make_mh_options(options), ) @@ -7472,16 +7491,19 @@ def _make_msg_has(self, message: 'Optional[Data_HomeAgentSwitchMessage]' = None, addresses: Alternate home agent addresses. An empty list is valid and asks the mobile node to run home agent discovery. options: Mobility options. - msg_reserved: The reserved octet of the message body as captured, - truncated or zero-filled to 1 octet. + msg_reserved: The reserved octet of the message body as captured. **kwargs: Arbitrary keyword arguments. Returns: Constructed message type. + Raises: + ProtocolError: If ``msg_reserved`` is not one octet. + """ if message is not None: - msg_reserved = getattr(message, 'msg_reserved', b'\x00') + msg_reserved = self._fill_short_read(getattr(message, 'msg_reserved', b'\x00'), + size=1, model=message, field='data', offset=1) addresses = cast('list[IPv6Address | bytes | str | int]', list(message.addresses)) options = message.options else: @@ -7490,7 +7512,7 @@ def _make_msg_has(self, message: 'Optional[Data_HomeAgentSwitchMessage]' = None, return Schema_HomeAgentSwitchMessage( count=len(addresses), - reserved=msg_reserved, + reserved=self._make_octets(msg_reserved, size=1, name='reserved octets'), addresses=addresses, options=self._make_mh_options(options), ) @@ -7624,8 +7646,7 @@ def _make_msg_lri(self, message: 'Optional[Data_LocalizedRoutingInitiationMessag one counts seconds rather than units of 4 seconds, so the value is not scaled. options: Mobility options. - msg_reserved: The two reserved octets of the message body as captured, - truncated or zero-filled to 2 octets. + msg_reserved: The two reserved octets of the message body as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -7633,10 +7654,12 @@ def _make_msg_lri(self, message: 'Optional[Data_LocalizedRoutingInitiationMessag Raises: ProtocolError: If ``seq`` or the lifetime does not fit its field. + Also if ``msg_reserved`` is not two octets. """ if message is not None: - msg_reserved = getattr(message, 'msg_reserved', b'\x00\x00') + msg_reserved = self._fill_short_read(getattr(message, 'msg_reserved', b'\x00\x00'), + size=2, model=message, field='data', offset=2) seq = message.seq lifetime_val = math.ceil(message.lifetime.total_seconds()) options = message.options @@ -7646,7 +7669,7 @@ def _make_msg_lri(self, message: 'Optional[Data_LocalizedRoutingInitiationMessag return Schema_LocalizedRoutingInitiationMessage( seq=self._make_uint(seq, bits=16, name='sequence number'), - reserved=msg_reserved, + reserved=self._make_octets(msg_reserved, size=2, name='reserved octets'), lifetime=self._make_uint(lifetime_val, bits=16, name='lifetime'), options=self._make_mh_options(options), ) @@ -7770,8 +7793,7 @@ def _make_msg_upa(self, message: 'Optional[Data_UpdateNotificationAcknowledgemen seq: Sequence number. status: Status. options: Mobility options. - msg_reserved: The three reserved octets of the message body as captured, - truncated or zero-filled to 3 octets. + msg_reserved: The three reserved octets of the message body as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -7780,10 +7802,12 @@ def _make_msg_upa(self, message: 'Optional[Data_UpdateNotificationAcknowledgemen Raises: ProtocolError: If ``seq`` does not fit its field, or ``status`` is a name, a :obj:`bool` or a code its field cannot carry. + Also if ``msg_reserved`` is not three octets. """ if message is not None: - msg_reserved = getattr(message, 'msg_reserved', b'\x00\x00\x00') + msg_reserved = self._fill_short_read(getattr(message, 'msg_reserved', b'\x00\x00\x00'), + size=3, model=message, field='data', offset=3) seq = message.seq status_val = message.status # type: Enum_UpdateNotificationACKStatus options = message.options @@ -7795,7 +7819,7 @@ def _make_msg_upa(self, message: 'Optional[Data_UpdateNotificationAcknowledgemen return Schema_UpdateNotificationAcknowledgementMessage( seq=self._make_uint(seq, bits=16, name='sequence number'), status=status_val, - reserved=msg_reserved, + reserved=self._make_octets(msg_reserved, size=3, name='reserved octets'), options=self._make_mh_options(options), ) @@ -7871,8 +7895,7 @@ def _make_msg_sq(self, message: 'Optional[Data_SubscriptionQueryMessage]' = None message: Message data model. seq: Sequence number, counted modulo 256. options: Mobility options. - msg_reserved: The reserved octet of the message body as captured, - truncated or zero-filled to 1 octet. + msg_reserved: The reserved octet of the message body as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -7880,10 +7903,12 @@ def _make_msg_sq(self, message: 'Optional[Data_SubscriptionQueryMessage]' = None Raises: ProtocolError: If ``seq`` does not fit its field. + Also if ``msg_reserved`` is not one octet. """ if message is not None: - msg_reserved = getattr(message, 'msg_reserved', b'\x00') + msg_reserved = self._fill_short_read(getattr(message, 'msg_reserved', b'\x00'), + size=1, model=message, field='data', offset=1) seq = message.seq options = message.options else: @@ -7891,7 +7916,7 @@ def _make_msg_sq(self, message: 'Optional[Data_SubscriptionQueryMessage]' = None return Schema_SubscriptionQueryMessage( seq=self._make_uint(seq, bits=8, name='sequence number'), - reserved=msg_reserved, + reserved=self._make_octets(msg_reserved, size=1, name='reserved octets'), options=self._make_mh_options(options), ) @@ -8245,8 +8270,7 @@ def _make_opt_mnp(self, type: 'Enum_Option', option: 'Optional[Data_MobileNetwor bits past ``prefix_length`` are written as given. prefix_length: Prefix length; defaults to that of a network or of text ending in ``/``, and to 128 for an address. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -8256,6 +8280,7 @@ def _make_opt_mnp(self, type: 'Enum_Option', option: 'Optional[Data_MobileNetwor ProtocolError: If ``prefix`` is not an IPv6 address or network, or ``prefix_length`` is negative, a :obj:`bool`, or does not fit in one octet. + Also if ``reserved`` is not one octet. Note: The prefix length is checked against its 8-bit field alone, as @@ -8292,7 +8317,7 @@ def _make_opt_mnp(self, type: 'Enum_Option', option: 'Optional[Data_MobileNetwor return Schema_MobileNetworkPrefixOption( type=type, length=18, - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), prefix_length=prefix_length, prefix=prefix_addr, ) @@ -8787,6 +8812,9 @@ def _make_opt_ct(self, type: 'Enum_Option', option: 'Optional[Data_CareofTestOpt Returns: Constructed option schema. + Raises: + ProtocolError: If ``token`` is not eight octets. + """ if option is not None: token = option.token @@ -8794,7 +8822,7 @@ def _make_opt_ct(self, type: 'Enum_Option', option: 'Optional[Data_CareofTestOpt return Schema_CareofTestOption( type=type, length=8, - token=token, + token=self._make_octets(token, size=8, name='token'), ) def _make_opt_exp(self, type: 'Enum_Option', option: 'Optional[Data_ExperimentalMobilityOption]' = None, *, @@ -9045,8 +9073,7 @@ def _make_opt_hnp(self, type: 'Enum_Option', option: 'Optional[Data_HomeNetworkP option: Option data model. prefix_length: Prefix length. prefix: Home network prefix. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -9054,7 +9081,7 @@ def _make_opt_hnp(self, type: 'Enum_Option', option: 'Optional[Data_HomeNetworkP Raises: ProtocolError: If ``prefix_length`` is negative, a :obj:`bool`, - or does not fit in one octet. + or does not fit in one octet. Also if ``reserved`` is not one octet. Note: The prefix length is checked against its 8-bit field alone, as @@ -9078,7 +9105,7 @@ def _make_opt_hnp(self, type: 'Enum_Option', option: 'Optional[Data_HomeNetworkP return Schema_HomeNetworkPrefixOption( type=type, length=18, - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), prefix_length=prefix_length, prefix=prefix, ) @@ -9129,8 +9156,7 @@ def _make_opt_att(self, type: 'Enum_Option', option: 'Optional[Data_AccessTechno type: Option type. option: Option data model. att: Access technology type. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -9139,6 +9165,7 @@ def _make_opt_att(self, type: 'Enum_Option', option: 'Optional[Data_AccessTechno Raises: ProtocolError: If ``att`` is a name, a :obj:`bool` or a code its field cannot carry. + Also if ``reserved`` is not one octet. """ if option is not None: @@ -9151,7 +9178,7 @@ def _make_opt_att(self, type: 'Enum_Option', option: 'Optional[Data_AccessTechno return Schema_AccessTechnologyTypeOption( type=type, length=2, - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), att=att_val, ) @@ -9165,8 +9192,7 @@ def _make_opt_mn_lli(self, type: 'Enum_Option', option: 'Optional[Data_MNLLIdent type: Option type. option: Option data model. lli: Link-layer identifier. - reserved: The two reserved octets as captured, truncated or zero-filled - to 2 octets. + reserved: The two reserved octets as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -9174,7 +9200,7 @@ def _make_opt_mn_lli(self, type: 'Enum_Option', option: 'Optional[Data_MNLLIdent Raises: ProtocolError: If the option is too long for its one-octet - ``Option Length``. + ``Option Length``. Also if ``reserved`` is not two octets. """ if option is not None: @@ -9184,7 +9210,7 @@ def _make_opt_mn_lli(self, type: 'Enum_Option', option: 'Optional[Data_MNLLIdent return Schema_MNLLIdentifierOption( type=type, length=self._make_uint(2 + len(lli), bits=8, name='Option Length'), - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets'), lli=lli, ) @@ -9437,13 +9463,15 @@ def _make_opt_ipv4_coa(self, type: 'Enum_Option', option: 'Optional[Data_IPv4Car type: Option type. option: Option data model. address: IPv4 care-of address. - reserved: The two reserved octets as captured, truncated or zero-filled - to 2 octets. + reserved: The two reserved octets as captured. **kwargs: Arbitrary keyword arguments. Returns: Constructed option schema. + Raises: + ProtocolError: If ``reserved`` is not two octets. + """ if option is not None: reserved = getattr(option, 'reserved', b'\x00\x00') @@ -9452,7 +9480,7 @@ def _make_opt_ipv4_coa(self, type: 'Enum_Option', option: 'Optional[Data_IPv4Car return Schema_IPv4CareofAddressOption( type=type, length=6, - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets'), address=address, ) @@ -9467,8 +9495,7 @@ def _make_opt_gre(self, type: 'Enum_Option', option: 'Optional[Data_GREKeyOption option: Option data model. key: GRE key identifier, or :obj:`None` to omit it -- which is what an option length of 2 means on the wire. - reserved: The two reserved octets as captured, truncated or zero-filled - to 2 octets. + reserved: The two reserved octets as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -9476,6 +9503,7 @@ def _make_opt_gre(self, type: 'Enum_Option', option: 'Optional[Data_GREKeyOption Raises: ProtocolError: If ``key`` does not fit its field. + Also if ``reserved`` is not two octets. """ if option is not None: @@ -9485,7 +9513,7 @@ def _make_opt_gre(self, type: 'Enum_Option', option: 'Optional[Data_GREKeyOption return Schema_GREKeyOption( type=type, length=2 if key is None else 6, - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets'), key=None if key is None else self._make_uint(key, bits=32, name='GRE key'), ) @@ -9680,13 +9708,15 @@ def _make_opt_ipv4_router(self, type: 'Enum_Option', type: Option type. option: Option data model. address: IPv4 default-router address. - reserved: The two reserved octets as captured, truncated or zero-filled - to 2 octets. + reserved: The two reserved octets as captured. **kwargs: Arbitrary keyword arguments. Returns: Constructed option schema. + Raises: + ProtocolError: If ``reserved`` is not two octets. + """ if option is not None: reserved = getattr(option, 'reserved', b'\x00\x00') @@ -9695,7 +9725,7 @@ def _make_opt_ipv4_router(self, type: 'Enum_Option', return Schema_IPv4DefaultRouterAddressOption( type=type, length=6, - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets'), address=address, ) @@ -9745,8 +9775,7 @@ def _make_opt_cr(self, type: 'Enum_Option', option: 'Optional[Data_ContextReques option: Option data model. requests: Requested contexts, each either a data model or a ``(mobility option type, extra data)`` pair. - reserved: The two reserved octets as captured, truncated or zero-filled - to 2 octets. + reserved: The two reserved octets as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -9756,6 +9785,7 @@ def _make_opt_cr(self, type: 'Enum_Option', option: 'Optional[Data_ContextReques ProtocolError: If the option is too long for its one-octet ``Option Length``, or the mobility option type of a pair is a name, a :obj:`bool` or a code its field cannot carry. + Also if ``reserved`` is not two octets. """ if option is not None: @@ -9784,7 +9814,7 @@ def _make_opt_cr(self, type: 'Enum_Option', option: 'Optional[Data_ContextReques return Schema_ContextRequestOption( type=type, length=self._make_uint(2 + len(buffer), bits=8, name='Option Length'), - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets'), requests=bytes(buffer), ) @@ -9802,8 +9832,7 @@ def _make_opt_lmaa(self, type: 'Enum_Option', option: 'Optional[Data_LMAAddressO :class:`~pcapkit.protocols.internet.mh.LMAAddressCode`. If not given, the code of the address family of ``address``. address: Local mobility anchor address. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -9814,6 +9843,7 @@ def _make_opt_lmaa(self, type: 'Enum_Option', option: 'Optional[Data_LMAAddressO :rfc:`5949#section-6.2.2` gives the address family of ``address`` (:issue:`2280`), or is a name, a :obj:`bool` or a code its field cannot carry. + Also if ``reserved`` is not one octet. Note: The option length is derived from the address family rather than from @@ -9860,7 +9890,7 @@ def _make_opt_lmaa(self, type: 'Enum_Option', option: 'Optional[Data_LMAAddressO type=type, length=6 if addr.version == 4 else 18, code=code_val, - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), address=addr, ) @@ -9874,13 +9904,16 @@ def _make_opt_mn_lla_iid(self, type: 'Enum_Option', option: 'Optional[Data_MNLLA type: Option type. option: Option data model. iid: Interface identifier; exactly 8 octets. - reserved: The two reserved octets as captured, truncated or zero-filled - to 2 octets. + reserved: The two reserved octets as captured. **kwargs: Arbitrary keyword arguments. Returns: Constructed option schema. + Raises: + ProtocolError: If ``iid`` is not eight octets, or ``reserved`` is not + two. + """ if option is not None: reserved = getattr(option, 'reserved', b'\x00\x00') @@ -9892,7 +9925,7 @@ def _make_opt_mn_lla_iid(self, type: 'Enum_Option', option: 'Optional[Data_MNLLA return Schema_MNLLAIIDOption( type=type, length=10, - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets'), iid=iid, ) @@ -10020,8 +10053,9 @@ def _make_fid_suboption(self, code: 'Enum_FlowIDSuboption', Raises: ProtocolError: If an entry of a ``BID_Reference`` sub-option's ``bid`` does - not fit its field, or a ``ts_format`` or ``action`` is a name, a - :obj:`bool` or a code its field cannot carry. + not fit its field, a ``ts_format`` or ``action`` is a name, a + :obj:`bool` or a code its field cannot carry, or a sub-option's + ``reserved`` is not as many octets as its field. Note: The data model parameter is named ``option`` rather than ``data`` @@ -10074,7 +10108,8 @@ def _make_fid_suboption(self, code: 'Enum_FlowIDSuboption', reserved = cast('bytes', kwargs.get('reserved', b'\x00')) return Schema_TrafficSelectorSuboption( type=code, length=2 + len(selector), - ts_format=cast('Enum_TrafficSelector', ts_format), reserved=reserved, + ts_format=cast('Enum_TrafficSelector', ts_format), + reserved=self._make_octets(reserved, size=1, name='reserved octets'), selector=selector) if code == Enum_FlowIDSuboption.Flow_Binding_Action: @@ -10087,7 +10122,7 @@ def _make_fid_suboption(self, code: 'Enum_FlowIDSuboption', Enum_FlowBindingAction, bits=8, name='action') reserved = cast('bytes', kwargs.get('reserved', b'\x00')) return Schema_FlowBindingActionSuboption( - type=code, length=2, reserved=reserved, + type=code, length=2, reserved=self._make_octets(reserved, size=1, name='reserved octets'), action=cast('Enum_FlowBindingAction', action)) if code == Enum_FlowIDSuboption.Target_Care_of_Address: @@ -10104,7 +10139,8 @@ def _make_fid_suboption(self, code: 'Enum_FlowIDSuboption', addr = parse_ip_address( address, f'{self.alias}: [OptNo {code}] invalid target care-of address') return Schema_TargetCareofAddressSuboption( - type=code, length=6 if addr.version == 4 else 18, reserved=reserved, + type=code, length=6 if addr.version == 4 else 18, + reserved=self._make_octets(reserved, size=2, name='reserved octets'), address=addr) if option is not None: @@ -10130,8 +10166,7 @@ def _make_opt_fid(self, type: 'Enum_Option', option: 'Optional[Data_FlowIdentifi fid_pri: Flow priority. status: Status. suboptions: Sub-options. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -10142,6 +10177,7 @@ def _make_opt_fid(self, type: 'Enum_Option', option: 'Optional[Data_FlowIdentifi Also if ``status`` is a name, a :obj:`bool` or a code its field cannot carry, or the option is too long for its one-octet ``Option Length``. + Also if ``reserved`` is not one octet. """ if option is not None: @@ -10164,7 +10200,7 @@ def _make_opt_fid(self, type: 'Enum_Option', option: 'Optional[Data_FlowIdentifi length=self._make_uint(length, bits=8, name='Option Length'), fid=self._make_uint(fid, bits=16, name='flow identifier'), fid_pri=self._make_uint(fid_pri, bits=16, name='flow identifier priority'), - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), status=status_val, suboptions=entries, ) @@ -10177,13 +10213,15 @@ def _make_opt_rc(self, type: 'Enum_Option', option: 'Optional[Data_RedirectCapab Args: type: Option type. option: Option data model. - reserved: The two reserved octets as captured, truncated or zero-filled - to 2 octets. + reserved: The two reserved octets as captured. **kwargs: Arbitrary keyword arguments. Returns: Constructed option schema. + Raises: + ProtocolError: If ``reserved`` is not two octets. + """ if option is not None: reserved = getattr(option, 'reserved', b'\x00\x00') @@ -10191,7 +10229,7 @@ def _make_opt_rc(self, type: 'Enum_Option', option: 'Optional[Data_RedirectCapab return Schema_RedirectCapabilityOption( type=type, length=2, - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets'), ) def _make_opt_redirect(self, type: 'Enum_Option', option: 'Optional[Data_RedirectOption]' = None, *, @@ -10320,8 +10358,7 @@ def _make_opt_mn_group(self, type: 'Enum_Option', option: 'Optional[Data_MNGroup option: Option data model. subtype: Sub-type. group_id: Mobile node group identifier. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -10331,6 +10368,7 @@ def _make_opt_mn_group(self, type: 'Enum_Option', option: 'Optional[Data_MNGroup ProtocolError: If ``group_id`` does not fit its field. Also if ``subtype`` is a name, a :obj:`bool` or a code its field cannot carry. + Also if ``reserved`` is not one octet. """ if option is not None: @@ -10346,7 +10384,7 @@ def _make_opt_mn_group(self, type: 'Enum_Option', option: 'Optional[Data_MNGroup type=type, length=6, subtype=subtype_val, - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), group_id=self._make_uint(group_id, bits=32, name='group identifier'), ) @@ -10363,8 +10401,7 @@ def _make_opt_mag_addr(self, type: 'Enum_Option', option: 'Optional[Data_MAGIPv6 address_length: Address length, in bits; :rfc:`6705#section-11.1` requires 128. address: MAG IPv6 address. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -10372,6 +10409,7 @@ def _make_opt_mag_addr(self, type: 'Enum_Option', option: 'Optional[Data_MAGIPv6 Raises: ProtocolError: If ``address_length`` does not fit its field. + Also if ``reserved`` is not one octet. """ if option is not None: @@ -10382,7 +10420,7 @@ def _make_opt_mag_addr(self, type: 'Enum_Option', option: 'Optional[Data_MAGIPv6 return Schema_MAGIPv6AddressOption( type=type, length=18, - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), address_length=self._make_uint(address_length, bits=8, name='address length'), address=address, ) @@ -10509,7 +10547,8 @@ def _make_ani_suboption(self, code: 'Enum_ANISuboption', octets = cast('bytes', kwargs.get('reserved', b'\x00')) return Schema_ANICivicLocationSuboption( type=code, length=2 + len(location), - format=self._make_uint(fmt, bits=8, name='civic location format'), reserved=octets, + format=self._make_uint(fmt, bits=8, name='civic location format'), + reserved=self._make_octets(octets, size=1, name='reserved octets'), location=location) if code == Enum_ANISuboption.MAG_Group_Identifier: @@ -10886,7 +10925,8 @@ def _make_qos_attribute(self, code: 'Enum_QoSAttribute', else: rate = cast('int', kwargs.get('rate', 0)) octets = cast('bytes', kwargs.get('reserved', b'\x00\x00')) - return Schema_BitRateAttribute(type=code, length=6, reserved=octets, + return Schema_BitRateAttribute(type=code, length=6, + reserved=self._make_octets(octets, size=2, name='reserved octets'), rate=self._make_uint(rate, bits=32, name='bit rate')) if code == Enum_QoSAttribute.Allocation_Retention_Priority: @@ -10902,7 +10942,7 @@ def _make_qos_attribute(self, code: 'Enum_QoSAttribute', vulnerability = cast('int', kwargs.get('preemption_vulnerability', 0)) octets = cast('bytes', kwargs.get('reserved', b'\x00')) return Schema_AllocationRetentionPriorityAttribute( - type=code, length=2, reserved=octets, + type=code, length=2, reserved=self._make_octets(octets, size=1, name='reserved octets'), priority={ 'PL': self._make_uint(level, bits=4, name='priority level'), 'PC': self._make_uint(capability, bits=2, name='preemption capability'), @@ -10922,7 +10962,8 @@ def _make_qos_attribute(self, code: 'Enum_QoSAttribute', selector = cast('bytes', kwargs.get('selector', b'')) octets = cast('bytes', kwargs.get('reserved', b'\x00')) return Schema_QoSTrafficSelectorAttribute( - type=code, length=2 + len(selector), reserved=octets, + type=code, length=2 + len(selector), + reserved=self._make_octets(octets, size=1, name='reserved octets'), ts_format=cast('Enum_TrafficSelector', ts_format), selector=selector) if code == Enum_QoSAttribute.QoS_Vendor_Specific_Attribute: @@ -10937,7 +10978,8 @@ def _make_qos_attribute(self, code: 'Enum_QoSAttribute', payload = cast('bytes', kwargs.get('data', b'')) octets = cast('bytes', kwargs.get('reserved', b'\x00\x00')) return Schema_QoSVendorSpecificAttribute( - type=code, length=7 + len(payload), reserved=octets, + type=code, length=7 + len(payload), + reserved=self._make_octets(octets, size=2, name='reserved octets'), vendor=self._make_uint(vendor, bits=32, name='vendor ID'), subtype=self._make_uint(subtype, bits=8, name='vendor subtype'), data=payload) @@ -10964,8 +11006,7 @@ def _make_opt_qos(self, type: 'Enum_Option', option: 'Optional[Data_QualityOfSer dscp: Differentiated services code point, 6 bits. oc: Operational code. attributes: Quality-of-service attributes. - reserved: The three reserved octets as captured, truncated or zero- - filled to 3 octets. + reserved: The three reserved octets as captured. tc_reserved: The 2 reserved bits of the traffic class, after ``dscp``. **kwargs: Arbitrary keyword arguments. @@ -10976,6 +11017,7 @@ def _make_opt_qos(self, type: 'Enum_Option', option: 'Optional[Data_QualityOfSer ProtocolError: If ``sr_id``, ``dscp``, ``tc_reserved`` or ``oc`` does not fit its field. Also if the option is too long for its one-octet ``Option Length``. + Also if ``reserved`` is not three octets. .. versionchanged:: 1.6.0 @@ -11006,7 +11048,7 @@ def _make_opt_qos(self, type: 'Enum_Option', option: 'Optional[Data_QualityOfSer sr_id=self._make_uint(sr_id, bits=8, name='service request identifier'), tc=dscp << 2 | tc_reserved, oc=self._make_uint(oc, bits=8, name='operational code'), - reserved=reserved, + reserved=self._make_octets(reserved, size=3, name='reserved octets'), attributes=entries, ) @@ -11022,18 +11064,21 @@ def _make_opt_lma_up(self, type: 'Enum_Option', option: 'Optional[Data_LMAUserPl address: LMA user-plane address, or :obj:`None` to omit it -- which is how a mobile access gateway asks for a transport without naming an address. - reserved: The two reserved octets as captured, truncated or zero-filled - to 2 octets. + reserved: The two reserved octets as captured. **kwargs: Arbitrary keyword arguments. Returns: Constructed option schema. + Raises: + ProtocolError: If ``reserved`` is not two octets. + """ if option is not None: reserved = getattr(option, 'reserved', b'\x00\x00') address = option.address + reserved = self._make_octets(reserved, size=2, name='reserved octets') if address is None: return Schema_LMAUserPlaneAddressOption(type=type, length=2, reserved=reserved, address=b'') @@ -11065,8 +11110,7 @@ def _make_opt_mcast(self, type: 'Enum_Option', option: 'Optional[Data_MulticastM option: Option data model. code: Option code. data: MLD or IGMP report payload, opaque to this module. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -11078,6 +11122,7 @@ def _make_opt_mcast(self, type: 'Enum_Option', option: 'Optional[Data_MulticastM a payload that is not a multiple of 4 octets cannot be described by it at all. Also if ``code`` does not fit in one octet, or the option is too long for its one-octet ``Option Length``. + Also if ``reserved`` is not one octet. """ if option is not None: @@ -11093,7 +11138,7 @@ def _make_opt_mcast(self, type: 'Enum_Option', option: 'Optional[Data_MulticastM type=type, length=self._make_uint(len(data) // 4, bits=8, name='Option Length'), code=self._make_uint(code, bits=8, name='code'), - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), data=data, ) @@ -11367,8 +11412,7 @@ def _make_opt_mag_id(self, type: 'Enum_Option', option: 'Optional[Data_MAGIdenti option: Option data model. subtype: Sub-type, from the mobile node identifier subtype registry. identifier: Identifier. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -11378,6 +11422,7 @@ def _make_opt_mag_id(self, type: 'Enum_Option', option: 'Optional[Data_MAGIdenti ProtocolError: If ``subtype`` is a name, a :obj:`bool` or a code its field cannot carry, or the option is too long for its one-octet ``Option Length``. + Also if ``reserved`` is not one octet. """ if option is not None: @@ -11393,7 +11438,7 @@ def _make_opt_mag_id(self, type: 'Enum_Option', option: 'Optional[Data_MAGIdenti type=type, length=self._make_uint(2 + len(identifier), bits=8, name='Option Length'), subtype=subtype_val, - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), identifier=identifier, ) @@ -11409,8 +11454,7 @@ def _make_opt_anchored(self, type: 'Enum_Option', option: 'Optional[Data_Anchore option: Option data model. prefix_length: Prefix length. prefix: Anchored prefix. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -11418,7 +11462,7 @@ def _make_opt_anchored(self, type: 'Enum_Option', option: 'Optional[Data_Anchore Raises: ProtocolError: If ``prefix_length`` is negative, a :obj:`bool`, - or longer than the address. + or longer than the address. Also if ``reserved`` is not one octet. """ if option is not None: @@ -11433,7 +11477,7 @@ def _make_opt_anchored(self, type: 'Enum_Option', option: 'Optional[Data_Anchore return Schema_AnchoredPrefixOption( type=type, length=18, - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), prefix_length=prefix_length, prefix=prefix, ) @@ -11450,8 +11494,7 @@ def _make_opt_local_prefix(self, type: 'Enum_Option', option: 'Optional[Data_Loc option: Option data model. prefix_length: Prefix length. prefix: Local prefix. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -11459,7 +11502,7 @@ def _make_opt_local_prefix(self, type: 'Enum_Option', option: 'Optional[Data_Loc Raises: ProtocolError: If ``prefix_length`` is negative, a :obj:`bool`, - or longer than the address. + or longer than the address. Also if ``reserved`` is not one octet. """ if option is not None: @@ -11474,7 +11517,7 @@ def _make_opt_local_prefix(self, type: 'Enum_Option', option: 'Optional[Data_Loc return Schema_LocalPrefixOption( type=type, length=18, - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), prefix_length=prefix_length, prefix=prefix, ) @@ -11493,8 +11536,7 @@ def _make_opt_prev_maar(self, type: 'Enum_Option', option: 'Optional[Data_Previo prefix_length: Prefix length of ``prefix``. maar: Previous MAAR's global address. prefix: Mobile node's home network prefix. - reserved: The reserved octet as captured, truncated or zero-filled to 1 - octet. + reserved: The reserved octet as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -11502,7 +11544,7 @@ def _make_opt_prev_maar(self, type: 'Enum_Option', option: 'Optional[Data_Previo Raises: ProtocolError: If ``prefix_length`` is negative, a :obj:`bool`, - or longer than the address. + or longer than the address. Also if ``reserved`` is not one octet. """ if option is not None: @@ -11518,7 +11560,7 @@ def _make_opt_prev_maar(self, type: 'Enum_Option', option: 'Optional[Data_Previo return Schema_PreviousMAAROption( type=type, length=34, - reserved=reserved, + reserved=self._make_octets(reserved, size=1, name='reserved octets'), prefix_length=prefix_length, maar=maar, prefix=prefix, @@ -11582,8 +11624,7 @@ def _make_opt_dlif_lladdr(self, type: 'Enum_Option', option: 'Optional[Data_DLIF type: Option type. option: Option data model. lla: Distributed logical interface's link-layer address. - reserved: The two reserved octets as captured, truncated or zero-filled - to 2 octets. + reserved: The two reserved octets as captured. **kwargs: Arbitrary keyword arguments. Returns: @@ -11591,7 +11632,7 @@ def _make_opt_dlif_lladdr(self, type: 'Enum_Option', option: 'Optional[Data_DLIF Raises: ProtocolError: If the option is too long for its one-octet - ``Option Length``. + ``Option Length``. Also if ``reserved`` is not two octets. """ if option is not None: @@ -11601,7 +11642,7 @@ def _make_opt_dlif_lladdr(self, type: 'Enum_Option', option: 'Optional[Data_DLIF return Schema_DLIFLinkLayerAddressOption( type=type, length=self._make_uint(2 + len(lla), bits=8, name='Option Length'), - reserved=reserved, + reserved=self._make_octets(reserved, size=2, name='reserved octets'), lla=lla, ) diff --git a/pcapkit/protocols/misc/checksum/ip_pseudo.py b/pcapkit/protocols/misc/checksum/ip_pseudo.py index 8eea44da00..2e2b31a5f3 100644 --- a/pcapkit/protocols/misc/checksum/ip_pseudo.py +++ b/pcapkit/protocols/misc/checksum/ip_pseudo.py @@ -226,7 +226,8 @@ def make(self, dst: Destination address, of the same IP version as ``src``. protocol: Upper-layer protocol number. upper_length: Upper-layer packet length, header included. - zero: Reserved octets, truncated or zero-filled to the layout's width. + zero: Reserved octets, one for IPv4 and three for IPv6; empty for + zeros. **kwargs: Arbitrary keyword arguments. Returns: @@ -235,9 +236,10 @@ def make(self, Raises: ProtocolError: If ``ip`` is neither IPv4 nor IPv6, a value is given by neither the arguments nor ``ip``, a header of ``ip`` is - malformed or not parsed, or ``protocol`` is a name, a + malformed or not parsed, ``protocol`` is a name, a :obj:`bool`, or a member or code the 8-bit protocol field - cannot carry. + cannot carry, or ``zero`` is neither empty nor the layout's + width. FieldValueError: If ``src`` or ``dst`` is not a valid address, they differ in IP version, or ``upper_length`` does not fit the layout's length field. @@ -270,10 +272,9 @@ def make(self, f'out of range 0 to {limit:#x}') proto = self._make_index(protocol, Enum_TransType, bits=8, name='protocol') - # NOTE: Kept at the layout's width, so that a constructed data model - # matches the parsed one. + # NOTE: Empty is zeros at the layout's width; any other width is refused. width = 3 if is_v6 else 1 - zero = zero[:width].ljust(width, b'\x00') + zero = bytes(width) if zero == b'' else self._make_octets(zero, size=width, name='zero octets') if is_v6: return Schema_IP_Pseudo( diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index dd9547d939..19cbf7a5a2 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -6941,7 +6941,7 @@ def _make_secrets_wireguard(self, type: 'Enum_SecretsType', secrets: 'Optional[D ) def _make_secrets_zigbee_nwk(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_ZigBeeNWKKey]' = None, *, - nwk_key: 'bytes' = b'', + nwk_key: 'bytes' = bytes(16), pan_id: 'int' = 0, data: 'Optional[bytes]' = None, **kwargs: 'Any') -> 'Schema_ZigBeeNWKKey': @@ -6955,7 +6955,7 @@ def _make_secrets_zigbee_nwk(self, type: 'Enum_SecretsType', secrets: 'Optional[ (:issue:`1986`); otherwise its fields are. On a parsed model the octets win, so an edit to its fields alone is silently ignored. - nwk_key: NWK key. + nwk_key: NWK key, 16 octets. pan_id: PAN ID. data: Secrets Data octets, written back verbatim in place of ``nwk_key`` and ``pan_id``, as the octets of ``secrets`` are. @@ -6964,11 +6964,18 @@ def _make_secrets_zigbee_nwk(self, type: 'Enum_SecretsType', secrets: 'Optional[ Returns: Constructed secrets schema. + Raises: + ProtocolError: If ``nwk_key`` is not 16 octets. + .. versionchanged:: 1.6.0 - Takes the Secrets Data octets as ``data``, so that the :obj:`dict` - of a parsed model rebuilds the key as captured, as the model itself - does, rather than as its zero-filled fields (:issue:`2698`). + * Takes the Secrets Data octets as ``data``, so that the :obj:`dict` + of a parsed model rebuilds the key as captured, as the model itself + does, rather than as its zero-filled fields (:issue:`2698`). + * ``nwk_key`` of any other width, empty included, raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather than + being padded or cut to the 16-octet key, and defaults to 16 zero + octets where it defaulted to empty (:issue:`2676`). """ if secrets is not None: @@ -6980,12 +6987,12 @@ def _make_secrets_zigbee_nwk(self, type: 'Enum_SecretsType', secrets: 'Optional[ pan_id = secrets.pan_id return Schema_ZigBeeNWKKey( - key=nwk_key, + key=self._make_octets(nwk_key, size=16, name='[ZigBee NWK Key] key', alias='PCAP-NG'), panid=pan_id, ) def _make_secrets_zigbee_aps(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_ZigBeeAPSKey]' = None, *, - aps_key: 'bytes' = b'', + aps_key: 'bytes' = bytes(16), pan_id: 'int' = 0, short_address: 'int' = 0, data: 'Optional[bytes]' = None, @@ -6997,7 +7004,7 @@ def _make_secrets_zigbee_aps(self, type: 'Enum_SecretsType', secrets: 'Optional[ secrets: Secrets data model. The Secrets Data octets it was parsed from, if any, are written back verbatim, as for :meth:`_make_secrets_zigbee_nwk`; otherwise its fields are. - aps_key: APS key. + aps_key: APS key, 16 octets. pan_id: PAN ID. short_address: Short address. data: Secrets Data octets, written back verbatim in place of the @@ -7007,10 +7014,17 @@ def _make_secrets_zigbee_aps(self, type: 'Enum_SecretsType', secrets: 'Optional[ Returns: Constructed secrets schema. + Raises: + ProtocolError: If ``aps_key`` is not 16 octets. + .. versionchanged:: 1.6.0 - Takes the Secrets Data octets as ``data``, as - :meth:`_make_secrets_zigbee_nwk` does (:issue:`2698`). + * Takes the Secrets Data octets as ``data``, as + :meth:`_make_secrets_zigbee_nwk` does (:issue:`2698`). + * ``aps_key`` of any other width, empty included, raises + :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather than + being padded or cut to the 16-octet key, and defaults to 16 zero + octets where it defaulted to empty (:issue:`2676`). """ if secrets is not None: @@ -7023,7 +7037,7 @@ def _make_secrets_zigbee_aps(self, type: 'Enum_SecretsType', secrets: 'Optional[ short_address = secrets.short_address return Schema_ZigBeeAPSKey( - key=aps_key, + key=self._make_octets(aps_key, size=16, name='[ZigBee APS Key] key', alias='PCAP-NG'), panid=pan_id, addr_high=short_address >> 16, addr_low=short_address & 0xFFFF, diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index a2b6a05e40..23ae9f1cda 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -2251,6 +2251,52 @@ def _make_octets(self, value: 'Any', *, size: 'int', name: 'str', f'must be {size} {unit}, not {value!r}') return bytes(value) + @classmethod + def _fill_short_read(cls, value: 'Any', *, size: 'int', model: 'Any', field: 'str', + offset: 'int' = 0) -> 'Any': + """Zero-fill the octets of ``value`` that a capture cut short. + + A model parsed from a capture cut short records, in its + ``__short_read__``, the field of :attr:`__schema__` the capture ended + inside and the octets of it present. A padding field keeps only those + octets, where the other fields read the rest as zeros, so a value + read at or past the cut comes back short. Zero-filled, it rebuilds as + the rest of the header does, and the rebuild is cut back to what was + captured. A value of any other width is handed back for + :meth:`_make_octets` to refuse. + + Args: + value: the value taken from ``model`` + size: octets in the value's field + model: the data model ``value`` was taken from + field: the field of :attr:`__schema__` that holds ``value`` + offset: where ``value`` starts in ``field`` + + Returns: + ``value`` zero-filled to ``size`` where it is exactly the octets + the cut kept of it; otherwise ``value`` itself. + + .. versionadded:: 1.6.0 + + """ + if not isinstance(value, (bytes, bytearray)) or len(value) >= size: + return value + get = getattr(model, 'get', None) + record = get('__short_read__') if callable(get) else None + if record is None: + return value + name, count = record + fields = list(cls.__schema__.__fields__) + if name == field: + kept = min(max(count - offset, 0), size) + elif name in fields and field in fields and fields.index(name) < fields.index(field): + kept = 0 + else: + return value + if len(value) == kept: + return bytes(value).ljust(size, b'\x00') + return value + def _make_uint(self, value: 'Any', *, bits: 'int', name: 'str', alias: 'Optional[str]' = None) -> 'int': """Check that ``value`` is an unsigned integer that fits in ``bits`` bits. diff --git a/tests/project/test_docs_value_helpers_2676_unit.py b/tests/project/test_docs_value_helpers_2676_unit.py new file mode 100644 index 0000000000..46af63ec6e --- /dev/null +++ b/tests/project/test_docs_value_helpers_2676_unit.py @@ -0,0 +1,59 @@ +# -*- coding: utf-8 -*- +"""Pins ``protocol.rst``'s entries for the shared value helpers (#2676). + +The helpers the protocols read and make their field values with -- among them +``_make_uint``, ``_make_sint``, ``_read_sint``, ``_uint_size`` and +``_make_octets`` -- are contract for a protocol's author, so each has an +``automethod`` on the ``Protocol`` autoclass and a place in the ``ProtocolBase`` +method group, linked to that entry. The page's "Value Helpers" section names +``pad_length`` beside them, by name only, as its module may move. + +Each helper is checked as a method of ``ProtocolBase`` with a docstring, read +from the source rather than imported, so this module runs as a project pin. + +""" + +from __future__ import annotations + +import ast +import pathlib +import re +import unittest + +ROOT = pathlib.Path(__file__).resolve().parents[2] + +#: The page that documents the helpers. +PAGE = ROOT / 'docs' / 'source' / 'pcapkit' / 'protocols' / 'protocol.rst' + +#: The module that defines them. +SOURCE = ROOT / 'pcapkit' / 'protocols' / 'protocol.py' + +#: The value helpers, in the page's order. +HELPERS = ('_read_unpack', '_read_octets', '_read_uint', '_read_sint', '_read_vector', '_read_whole_vector', + '_read_uints', '_make_pack', '_make_octets', '_fill_short_read', '_make_uint', '_make_sint', + '_uint_size', '_make_vector', '_make_uints') + + +class ValueHelperEntryTests(unittest.TestCase): + """Each helper has an entry on ``Protocol`` and a place in the ``ProtocolBase`` group.""" + + text = PAGE.read_text(encoding='utf-8') + + def test_helpers_are_documented(self) -> None: + tree = ast.parse(SOURCE.read_text(encoding='utf-8')) + cls = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == 'ProtocolBase') + methods = {node.name: node for node in cls.body if isinstance(node, ast.FunctionDef)} + protocol, base = self.text.split('.. autoclass:: pcapkit.protocols.protocol.ProtocolBase') + group = re.search(r'\.\. py:method:: id\n((?: {18}\w+\n)+)', base) + assert group is not None + for name in HELPERS: + with self.subTest(helper=name): + self.assertTrue(ast.get_docstring(methods[name])) + self.assertIn(f' .. automethod:: {name}\n', protocol) + self.assertIn(f' {name}\n', group.group(1)) + self.assertIn(f':meth:`~Protocol.{name}`', base) + self.assertIn('pad_length`', protocol) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_hip_unit.py b/tests/protocols/internet/test_hip_unit.py index 591a9f02e2..0947332c3f 100644 --- a/tests/protocols/internet/test_hip_unit.py +++ b/tests/protocols/internet/test_hip_unit.py @@ -1208,7 +1208,7 @@ def test_hip_parameter_constructors_cover_keyword_paths(self) -> None: Parameter.PAYLOAD_MIC, version=2, next=TransType.TCP, - payload=b'payload', + payload=b'payl', mic=b'mic', ).next, TransType.TCP) self.assertEqual(proto._make_param_transaction_id( @@ -1701,9 +1701,9 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> hip_data.PayloadMICParameter(type=Parameter.PAYLOAD_MIC, critical=False, reserved=bytes(3), length=16, next=TransType.TCP, - payload=b'payload', mic=b'mic'), + payload=b'payl', mic=b'mic'), version=2, - ).payload, b'payload') + ).payload, b'payl') self.assertEqual(proto._make_param_transaction_id( Parameter.TRANSACTION_ID, hip_data.TransactionIDParameter(type=Parameter.TRANSACTION_ID, diff --git a/tests/protocols/misc/checksum/test_ip_pseudo_2471_unit.py b/tests/protocols/misc/checksum/test_ip_pseudo_2471_unit.py index cb2f8ae8c3..5ecee9e9ef 100644 --- a/tests/protocols/misc/checksum/test_ip_pseudo_2471_unit.py +++ b/tests/protocols/misc/checksum/test_ip_pseudo_2471_unit.py @@ -376,17 +376,25 @@ def test_upper_length_range(self) -> None: with self.assertRaisesRegex(FieldValueError, 'length -20 out of range'): IP_Pseudo(ip=tcp) - def test_zero_is_kept_at_the_layout_width(self) -> None: + def test_zero_is_the_layout_width(self) -> None: + # One zero octet for IPv4 (RFC 793 section 3.1), three for IPv6 (RFC + # 8200 section 8.1); empty is zeros, and any other width is refused (#2676). from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo + from pcapkit.utilities.exceptions import ProtocolError - ipv4 = IP_Pseudo(src='10.0.0.1', dst='10.0.0.2', protocol=6, upper_length=20, zero=b'\xaa\xbb') - self.assertEqual(ipv4.data, bytes.fromhex('0a000001' '0a000002' 'aa' '06' '0014')) - self.assertEqual(ipv4.info.zero, b'\xaa') - for zero, kept in ((b'\xaa\xbb\xcc\xdd', 'aabbcc'), (b'\xaa', 'aa0000')): - with self.subTest(zero=zero): - ipv6 = IP_Pseudo(src='::1', dst='::2', protocol=6, upper_length=20, zero=zero) - self.assertEqual(ipv6.data[32:], bytes.fromhex('00000014') + bytes.fromhex(kept) + b'\x06') - self.assertEqual(ipv6.info.zero, bytes.fromhex(kept)) + ipv4 = IP_Pseudo(src='10.0.0.1', dst='10.0.0.2', protocol=6, upper_length=20, zero=b'\xff') + self.assertEqual(ipv4.data, bytes.fromhex('0a000001' '0a000002' 'ff' '06' '0014')) + self.assertEqual(ipv4.info.zero, b'\xff') + ipv6 = IP_Pseudo(src='::1', dst='::2', protocol=6, upper_length=20, zero=b'\xff' * 3) + self.assertEqual(ipv6.data[32:], bytes.fromhex('00000014' 'ffffff' '06')) + self.assertEqual(ipv6.info.zero, b'\xff' * 3) + self.assertEqual(IP_Pseudo(src='::1', dst='::2', protocol=6, upper_length=20).info.zero, bytes(3)) + + for src, dst, zero in (('10.0.0.1', '10.0.0.2', b'\xaa\xbb'), ('::1', '::2', b'\xaa\xbb\xcc\xdd'), + ('::1', '::2', b'\xaa'), ('10.0.0.1', '10.0.0.2', 1)): + with self.subTest(src=src, zero=zero): + with self.assertRaisesRegex(ProtocolError, r'^IP_Pseudo: invalid zero octets: must be \d octets?'): + IP_Pseudo(src=src, dst=dst, protocol=6, upper_length=20, zero=zero) def test_each_value_is_required_without_ip(self) -> None: from pcapkit.protocols.misc.checksum.ip_pseudo import IP_Pseudo diff --git a/tests/protocols/test_maker_octet_widths_2676_unit.py b/tests/protocols/test_maker_octet_widths_2676_unit.py new file mode 100644 index 0000000000..90c82fb141 --- /dev/null +++ b/tests/protocols/test_maker_octet_widths_2676_unit.py @@ -0,0 +1,764 @@ +# -*- coding: utf-8 -*- +"""The makers refuse a fixed-size octet field, or a timestamp count, that does not fit. + +GitHub issue #2676: a schema field of a fixed number of octets packs as +:mod:`struct`'s ``s`` format does, so a maker that hands it a caller's octets +unchecked pads a short value and cuts a long one, as the CALIPSO checksum did +with 300 octets. Each maker routes such a value through +:meth:`~pcapkit.protocols.protocol.Protocol._make_octets`, which raises a +:exc:`~pcapkit.utilities.exceptions.ProtocolError` for any other width, as #2617 +has every out-of-range value raise rather than be masked. + +* :class:`SweepTests` finds every fixed-size octet field a maker writes, from the + schema and protocol sources, so a new one is held to the rule without a row. +* The protocol classes check each field's refusal, and that its widest value + builds at the offset its specification gives and survives parse -> rebuild. + The MH and HIP rows come from the maker signatures. +* :class:`IPv4TimestampTests` covers the timestamp option's slot count, bounded by + the option's 40 octets, and its default, derived from the timestamps' form. +* :class:`CapturedOctetTests` checks that a capture cut inside a reserved field + still rebuilds as captured, and that a model of the wrong width is refused. + +``protocol.rst``'s entries for the value helpers are pinned in +:file:`tests/project/test_docs_value_helpers_2676_unit.py`. + +Every case builds its own octets in memory and reads no capture. + +""" + +import ast +import inspect +import io +import pathlib +import unittest +import warnings + +from tests._support import reimport_once_per_class + +ROOT = pathlib.Path(__file__).resolve().parents[2] + +#: Fixed-size octet fields a maker writes without ``_make_octets``, and why. +EXEMPT = { + ('pcapkit/protocols/application/httpv2.py', '_make_http_ping', 'data'): + 'its own length check refuses any width but eight', + ('pcapkit/protocols/internet/mh.py', '_make_opt_mn_lla_iid', 'iid'): + 'its own length check refuses any width but eight', + ('pcapkit/protocols/link/linux_sll1.py', 'make', 'link_addr'): + '_validate_link_addr refuses more than eight octets and zero-fills a shorter ' + 'address, whose length link_addr_length carries', + ('pcapkit/protocols/link/linux_sll2.py', 'make', 'link_addr'): + '_validate_link_addr refuses more than eight octets and zero-fills a shorter ' + 'address, whose length link_addr_length carries', + ('pcapkit/protocols/link/radiotap.py', '_make_tlv_data', 'oui'): + '_make_vendor_data refuses any width but the field\'s', + ('pcapkit/protocols/link/radiotap.py', '_make_vendor_namespace', 'oui'): + '_make_vendor_data refuses any width but the field\'s', + ('pcapkit/protocols/misc/pcap/header.py', 'make', 'magic_number'): + 'made from the byte order and resolution flags, not given as octets', +} + +#: Helpers whose result is exactly the width they are given. +CHECKED = ('_make_octets', '_make_mac_addr', '_make_pack') + + +def _base_name(node: 'ast.expr') -> 'str | None': + """The bare name a base class is written with, its subscript dropped: ``EnumSchema[X]`` is ``EnumSchema``.""" + if isinstance(node, ast.Subscript): + node = node.value + return node.id if isinstance(node, ast.Name) else None + + +def _schema_fields() -> 'dict[tuple[str, str], tuple[dict[str, int | None], list[tuple[str, str]]]]': + """``{(module, class): ({field: fixed size or None}, [base (module, class)])}`` for every schema.""" + classes = {} # type: dict[tuple[str, str], tuple[dict[str, int | None], list[tuple[str, str]]]] + for path in sorted((ROOT / 'pcapkit/protocols/schema').rglob('*.py')): + module = '.'.join(path.relative_to(ROOT).with_suffix('').parts) + tree = ast.parse(path.read_text(encoding='utf-8')) + names = {node.name: (module, node.name) for node in tree.body if isinstance(node, ast.ClassDef)} + for node in tree.body: + if isinstance(node, ast.ImportFrom) and node.module and node.module.startswith('pcapkit.protocols.schema'): + names.update({alias.asname or alias.name: (node.module, alias.name) for alias in node.names}) + for node in tree.body: + if not isinstance(node, ast.ClassDef): + continue + fields = {} # type: dict[str, int | None] + for stmt in node.body: + if isinstance(stmt, ast.AnnAssign) and isinstance(stmt.target, ast.Name): + size = None + call = stmt.value + if isinstance(call, ast.Call) and getattr(call.func, 'id', None) in ('BytesField', 'PaddingField'): + for kw in call.keywords: + if kw.arg == 'length' and isinstance(kw.value, ast.Constant) \ + and isinstance(kw.value.value, int): + size = kw.value.value + fields[stmt.target.id] = size + bases = [names[base] for base in map(_base_name, node.bases) if base in names] + classes[(module, node.name)] = (fields, bases) + return classes + + +def _fixed_size(classes: 'dict', key: 'tuple[str, str]', field: 'str') -> 'int | None': + """The fixed width of ``field`` on schema ``key``, from the nearest class that declares it.""" + if key not in classes: + return None + fields, bases = classes[key] + if field in fields: + return fields[field] + for base in bases: + size = _fixed_size(classes, base, field) + if size is not None: + return size + return None + + +def _checked(node: 'ast.expr', size: 'int', assigned: 'dict[str, list[tuple[int, ast.expr]]]', line: 'int') -> 'bool': + """Whether ``node`` is exactly ``size`` octets by construction.""" + if isinstance(node, ast.Constant): + return isinstance(node.value, bytes) and len(node.value) == size + if isinstance(node, ast.Call): + return isinstance(node.func, ast.Attribute) and node.func.attr in CHECKED + if isinstance(node, ast.IfExp): + return _checked(node.body, size, assigned, line) and _checked(node.orelse, size, assigned, line) + if isinstance(node, ast.Name): + before = [value for lineno, value in assigned.get(node.id, []) if lineno < line] + return bool(before) and _checked(before[-1], size, assigned, line) + return False + + +def unchecked_octet_fields() -> 'tuple[set[tuple[str, str, str]], set[tuple[str, str, str]]]': + """Every fixed-size octet field a maker writes, and those it writes unchecked: ``(path, maker, field)``.""" + classes = _schema_fields() + found, unchecked = set(), set() + for path in sorted((ROOT / 'pcapkit/protocols').rglob('*.py')): + if 'schema' in path.parts or 'data' in path.parts: + continue + rel = path.relative_to(ROOT).as_posix() + tree = ast.parse(path.read_text(encoding='utf-8')) + imports = {} + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module and node.module.startswith('pcapkit.protocols.schema'): + imports.update({alias.asname or alias.name: (node.module, alias.name) for alias in node.names}) + for func in ast.walk(tree): + if not isinstance(func, ast.FunctionDef) or not (func.name == 'make' or func.name.startswith('_make')): + continue + assigned = {} # type: dict[str, list[tuple[int, ast.expr]]] + for node in ast.walk(func): + if isinstance(node, ast.Assign): + for target in node.targets: + if isinstance(target, ast.Name): + assigned.setdefault(target.id, []).append((node.lineno, node.value)) + for value in assigned.values(): + value.sort(key=lambda item: item[0]) + for node in ast.walk(func): + if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in imports): + continue + for kw in node.keywords: + size = _fixed_size(classes, imports[node.func.id], kw.arg) if kw.arg else None + if size is None: + continue + found.add((rel, func.name, kw.arg)) + if not _checked(kw.value, size, assigned, node.lineno): + unchecked.add((rel, func.name, kw.arg)) + return found, unchecked + + +def _octet_arguments(func: 'object', exclude: 'set[str]') -> 'list[tuple[str, int]]': + """``(argument, width)`` for each keyword of ``func`` that defaults to fixed-size octets.""" + return [(param.name, len(param.default)) + for param in inspect.signature(func).parameters.values() # type: ignore[arg-type] + if str(param.annotation) == 'bytes' and isinstance(param.default, bytes) and param.default + and param.name not in exclude] + + +class _Case(unittest.TestCase): + """Fresh :mod:`pcapkit` once per class, and the octet-width checks.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def assertWidth(self, alias: 'str', name: 'str', size: 'int', build: 'object', rebuild: 'object', + offset: 'int | None' = None, where: 'str' = '') -> None: + """``build(octets)`` refuses one octet too few or too many, and builds and rebuilds ``size`` of ``0xFF``.""" + from pcapkit.utilities.exceptions import ProtocolError + + unit = 'octet' if size == 1 else 'octets' + for value in (b'\xff' * (size - 1), b'\xff' * (size + 1)): + with self.subTest(where=where, field=name, width=len(value)): + with self.assertRaises(ProtocolError) as caught: + build(value) # type: ignore[operator] + self.assertEqual(str(caught.exception), f'{alias}: invalid {name}: must be {size} {unit}, ' + f'not {value!r}') + with self.subTest(where=where, field=name, width=size): + data = build(b'\xff' * size) # type: ignore[operator] + if offset is None: + self.assertIn(b'\xff' * size, data) + else: + self.assertEqual(data[offset:offset + size], b'\xff' * size) + self.assertEqual(rebuild(data), data) # type: ignore[operator] + + +class SweepTests(unittest.TestCase): + """Every fixed-size octet field a maker writes is checked for its width.""" + + def test_every_fixed_octet_field_is_checked(self) -> None: + # An exemption names a field the sweep finds unchecked, so a fix + # deletes its row. + found, unchecked = unchecked_octet_fields() + self.assertIn(('pcapkit/protocols/internet/hopopt.py', '_make_opt_calipso', 'checksum'), found) + self.assertEqual(sorted(unchecked - set(EXEMPT)), []) + self.assertEqual(sorted(set(EXEMPT) - unchecked), []) + + +class IPv6OptionTests(_Case): + """HOPOPT and IPv6-Opts, which share their option makers.""" + + def _classes(self) -> 'dict[str, type]': + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts + return {'HOPOPT': HOPOPT, 'IPv6-Opts': IPv6_Opts} + + def test_calipso_checksum(self) -> None: + # RFC 5570 section 5.1: type, length, the 32-bit DOI, Cmpt Length and + # Sens Level, then the 16-bit checksum, after the 2-octet header. + from pcapkit.const.ipv6.option import Option + + for alias, cls in self._classes().items(): + self.assertWidth(alias, 'CALIPSO checksum', 2, + lambda value, cls=cls: cls(next=59, options=[(Option.CALIPSO, {'checksum': value})]).data, + lambda data, cls=cls: cls.from_data(cls(data, len(data), extension=True).info).data, + offset=10) + + def test_quick_start_report_not_used_octet(self) -> None: + # RFC 4782 section 3.1, figure 4, which section 3.2 keeps for IPv6: the + # function and rate, then the Not Used octet. + from pcapkit.const.ipv6.option import Option + from pcapkit.const.ipv6.qs_function import QSFunction + + for alias, cls in self._classes().items(): + self.assertWidth(alias, 'Quick-Start not-used octet', 1, + lambda value, cls=cls: cls(next=59, options=[(Option.Quick_Start, { + 'func': QSFunction.Report_of_Approved_Rate, 'unused': value})]).data, + lambda data, cls=cls: cls.from_data(cls(data, len(data), extension=True).info).data, + offset=5) + + +class IPv4TimestampTests(_Case): + """The timestamp option's slot count, and IPv4's header checksum.""" + + def _make(self, **args: 'object') -> 'bytes': + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + + return IPv4(protocol=253, options=[(OptionNumber.TS, args)]).data + + def _rebuild(self, data: 'bytes') -> 'bytes': + from pcapkit.protocols.internet.ipv4 import IPv4 + + return IPv4.from_data(IPv4(data).info).data + + def test_count_is_what_the_option_holds(self) -> None: + # RFC 791 section 3.1: the option is 40 octets at most, so nine 4-octet + # timestamps, or four 8-octet address and timestamp pairs. A count + # past that is refused; the widest fills the option, and the default + # is five timestamps, or four pairs. The option's length octet + # follows its type, at octet 21 of the header. + from pcapkit.utilities.exceptions import ProtocolError + + for timestamp, counts in (([], 10), ({}, 5), ([], 1 << 64), ({}, 9)): + with self.subTest(timestamp=type(timestamp).__name__, counts=counts): + with self.assertRaises(ProtocolError) as caught: + self._make(counts=counts, timestamp=timestamp) + self.assertEqual(str(caught.exception), f'IPv4: invalid timestamp count: {counts}') + for timestamp, counts, length in (([], 9, 40), ({}, 4, 36), ([], None, 24), ({}, None, 36), + (None, None, 24)): + with self.subTest(timestamp=type(timestamp).__name__, counts=counts): + data = self._make(timestamp=timestamp) if counts is None else \ + self._make(counts=counts, timestamp=timestamp) + self.assertEqual(data[21], length) + self.assertEqual(self._rebuild(data), data) + + def test_header_checksum(self) -> None: + # RFC 791 section 3.1: the Header Checksum is octets 10 and 11. + from pcapkit.protocols.internet.ipv4 import IPv4 + + self.assertWidth('IPv4', 'checksum', 2, lambda value: IPv4(protocol=253, checksum=value).data, + self._rebuild, offset=10) + + +class MHTests(_Case): + """MH's header, message, option and sub-option makers.""" + + #: Octet arguments with a non-empty default that are not of a fixed size, and why. + VARIABLE = { + # The message data of an Experimental Mobility Header, of any length. + '_make_msg_emh.data', + } + + #: Sub-options and attributes, which no maker signature declares: + #: ``(option, list argument, entry code, width)``; the entry takes ``reserved``. + NESTED = ( + ('Flow_Identification_Mobility_Option', 'suboptions', 'Traffic_Selector', 1), + ('Flow_Identification_Mobility_Option', 'suboptions', 'Flow_Binding_Action', 1), + ('Flow_Identification_Mobility_Option', 'suboptions', 'Target_Care_of_Address', 2), + ('Access_Network_Identifier', 'suboptions', 'Civic_Location', 1), + ('Quality_of_Service', 'attributes', 'Guaranteed_DL_Bit_Rate', 2), + ('Quality_of_Service', 'attributes', 'Allocation_Retention_Priority', 1), + ('Quality_of_Service', 'attributes', 'QoS_Traffic_Selector', 1), + ('Quality_of_Service', 'attributes', 'QoS_Vendor_Specific_Attribute', 2), + ) + + def _make(self, *options: 'tuple[str, dict[str, object]]', **kwargs: 'object') -> 'bytes': + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.protocols.internet.mh import MH + + return bytes(MH(next=59, type=Packet.Binding_Refresh_Request, + data={'options': [(Option[name], args) for name, args in options]}, **kwargs)) + + def _rebuild(self, data: 'bytes') -> 'bytes': + from pcapkit.protocols.internet.mh import MH + + return MH.from_data(MH(io.BytesIO(data), len(data), extension=True).info).data + + def test_header_checksum(self) -> None: + # RFC 6275 section 6.1.1: Checksum is octets 4 and 5. + self.assertWidth('MH', 'checksum', 2, lambda value: self._make(chksum=value), self._rebuild, offset=4) + + def test_message_octets(self) -> None: + from pcapkit.protocols.internet.mh import MH + + packets = {name: code for code, name in MH.__message__.items() if isinstance(name, str)} + swept = [] + for name in sorted(vars(MH)): + if not name.startswith('_make_msg_'): + continue + for arg, size in _octet_arguments(getattr(MH, name), {a.split('.')[1] for a in self.VARIABLE + if a.startswith(f'{name}.')}): + swept.append(f'{name}.{arg}') + code = packets[name[len('_make_msg_'):]] + label = 'reserved octets' if arg == 'msg_reserved' else arg + self.assertWidth('MH', label, size, + lambda value, code=code, arg=arg: bytes(MH(next=59, type=code, data={arg: value})), + self._rebuild, where=f'{name}.{arg}') + self.assertIn('_make_msg_hot.token', swept) + + def test_option_octets(self) -> None: + # The interface identifier is a fixed 8 octets (RFC 5949), + # which its maker refuses at any other width in a message of its own. + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + swept = [] + for code, name in sorted(MH.__option__.items()): + maker = f'_make_opt_{name}' + if not isinstance(name, str) or not hasattr(MH, maker) or maker in swept: + continue + swept.append(maker) + for arg, size in _octet_arguments(getattr(MH, maker), set()): + label = {'reserved': 'reserved octets'}.get(arg, arg) + if maker == '_make_opt_mn_lla_iid' and arg == 'iid': + for iid in (bytes(7), bytes(9)): + with self.subTest(where=f'{maker}.{arg}', width=len(iid)): + with self.assertRaisesRegex(ProtocolError, 'invalid interface identifier'): + self._make((Option(code).name, {arg: iid})) + continue + self.assertWidth('MH', label, size, + lambda value, code=code, arg=arg: self._make((Option(code).name, {arg: value})), + self._rebuild, where=f'{maker}.{arg}') + self.assertIn('_make_opt_ct', swept) + + def test_nested_octets(self) -> None: + from pcapkit.const.mh.ani_suboption import ANISuboption + from pcapkit.const.mh.flow_id_suboption import FlowIDSuboption + from pcapkit.const.mh.qos_attribute import QoSAttribute + + enums = (FlowIDSuboption, ANISuboption, QoSAttribute) + for option, key, entry, size in self.NESTED: + enum = next(e for e in enums if entry in e.__members__ and e.__name__ != 'Option') + with self.subTest(option=option, entry=entry): + self.assertWidth('MH', 'reserved octets', size, + lambda value, option=option, key=key, entry=enum[entry]: self._make( + (option, {key: [(entry, {'reserved': value})]})), + self._rebuild) + + +class HIPTests(_Case): + """HIP's header and parameter makers.""" + + #: Arguments a parameter needs to build at all, by maker suffix. + BASE = {'puzzle': {'lifetime': 1}} + + def _rebuild(self, data: 'bytes') -> 'bytes': + from pcapkit.protocols.internet.hip import HIP + from pcapkit.utilities.warnings import ProtocolWarning + + # An ENCRYPTED parameter with no HIP_CIPHER beside it warns on read. + with warnings.catch_warnings(): + warnings.simplefilter('ignore', ProtocolWarning) + return HIP.from_data(HIP(data).info).data + + def test_header_checksum(self) -> None: + # RFC 7401 section 5.1: Checksum is octets 4 and 5. + from pcapkit.protocols.internet.hip import HIP + + self.assertWidth('HIPv2', 'checksum', 2, lambda value: bytes(HIP(version=2, checksum=value)), + self._rebuild, offset=4) + + def test_parameter_octets(self) -> None: + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + + names = {'reserved': 'reserved octets', 'payload': 'payload data'} + swept = [] + for code, name in sorted(HIP.__parameter__.items()): + maker = f'_make_param_{name}' + # ``R1_COUNTER`` has a HIPv1-only code as well as its HIPv2 one. + if not isinstance(name, str) or not hasattr(HIP, maker) or maker in swept \ + or code == Parameter.R1_Counter: + continue + swept.append(maker) + for arg, size in _octet_arguments(getattr(HIP, maker), set()): + label = names.get(arg, f'{name} {arg}') + base = self.BASE.get(name, {}) + self.assertWidth('HIPv2', label, size, + lambda value, code=code, arg=arg, base=base: bytes(HIP( + version=2, parameters=[(code, {**base, arg: value})])), + self._rebuild, where=f'{maker}.{arg}') + self.assertIn('_make_param_payload_mic', swept) + + +class OtherMakerTests(_Case): + """The other protocols the sweep reaches.""" + + def test_icmp_checksum_and_unused_octets(self) -> None: + # RFC 792 and RFC 4443 sections 2.1 and 3: the checksum is octets 2 and + # 3, and the unused octets follow it, after a pointer octet for + # Parameter Problem. ICMPv4 takes RFC 792's name, ICMP, as its alias. + from pcapkit.protocols.internet.icmpv4 import ICMPv4 + from pcapkit.protocols.internet.icmpv6 import ICMPv6 + + for alias, cls, rows in (('ICMP', ICMPv4, ((3, 4, 4), (11, 4, 4), (12, 3, 5))), + ('ICMPv6', ICMPv6, ((1, 4, 4), (3, 4, 4)))): + rebuild = (lambda data, cls=cls: cls.from_data(cls(data).info).data) # noqa: E731 + self.assertWidth(alias, 'checksum', 2, lambda value, cls=cls: cls(type=8 if cls is ICMPv4 else 128, + checksum=value).data, + rebuild, offset=2) + for kind, size, offset in rows: + with self.subTest(protocol=alias, type=kind): + self.assertWidth(alias, 'unused octets', size, + lambda value, cls=cls, kind=kind: cls(type=kind, body={'unused': value}).data, + rebuild, offset=offset) + + def test_igmp_checksum_and_reserved(self) -> None: + # RFC 2236 section 2 and RFC 3376 section 4.2: the checksum is octets 2 + # and 3, and a version 3 report's second reserved field octets 4 and 5. + from pcapkit.const.igmp.type import Type + from pcapkit.protocols.internet.igmpv2 import IGMPv2 + from pcapkit.protocols.internet.igmpv3 import IGMPv3 + + self.assertWidth('IGMPv2', 'checksum', 2, + lambda value: IGMPv2(type=Type.IGMP_Membership_Query, max_resp_time=1, checksum=value).data, + lambda data: IGMPv2.from_data(IGMPv2(data).info).data, offset=2) + report = Type.IGMPv3_Membership_Report + for arg, name, offset in (('checksum', 'checksum', 2), ('reserved_2', 'reserved_2', 4)): + self.assertWidth('IGMPv3', name, 2, lambda value, arg=arg: IGMPv3(type=report, **{arg: value}).data, + lambda data: IGMPv3.from_data(IGMPv3(data).info).data, offset=offset) + + def test_ipv6_frag_and_route_reserved(self) -> None: + # RFC 8200 sections 4.5 and 4.4, and RFC 6275 section 6.4: the reserved + # octet follows Next Header; the reserved word follows Segments Left. + from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + self.assertWidth('IPv6-Frag', 'reserved octet', 1, + lambda value: IPv6_Frag(next=59, reserved_octet=value).data, + lambda data: IPv6_Frag.from_data(IPv6_Frag(data, len(data), extension=True).info).data, + offset=1) + for kind, data in ((0, {'ip': ['2001:db8::1']}), (2, {'ip': '2001:db8::1'})): + with self.subTest(type=kind): + self.assertWidth('IPv6-Route', 'reserved octets', 4, + lambda value, kind=kind, data=data: IPv6_Route( + next=59, type=kind, data={**data, 'reserved': value}).data, + lambda octets: IPv6_Route.from_data( + IPv6_Route(octets, len(octets), extension=True).info).data, + offset=4) + + def test_ipx_checksum_and_addresses(self) -> None: + # The IPX header: checksum at octet 0, the 12-octet destination and + # source addresses at octets 6 and 18. + from pcapkit.protocols.internet.ipx import IPX + + for arg, name, size, offset in (('chksum', 'checksum', 2, 0), ('dst', 'destination address', 12, 6), + ('src', 'source address', 12, 18)): + self.assertWidth('IPX', name, size, lambda value, arg=arg: IPX(**{arg: value}).data, + lambda data: IPX.from_data(IPX(data).info).data, offset=offset) + + def test_ospf_checksum_and_authentication_reserved(self) -> None: + # RFC 2328 appendix A.3.1 and D.3: the checksum is octets 12 and 13; + # cryptographic authentication opens with two reserved octets at 16. + from pcapkit.const.ospf.authentication import Authentication + from pcapkit.protocols.application.ospf import OSPF + + rebuild = (lambda data: OSPF.from_data(OSPF(data).info).data) # noqa: E731 + self.assertWidth('OSPF', 'checksum', 2, lambda value: OSPF(checksum=value).data, rebuild, offset=12) + auth = {'key_id': 1, 'len': 16, 'seq': 1} + self.assertWidth('OSPF', 'reserved octets', 2, + lambda value: OSPF(auth_type=Authentication.Cryptographic_authentication, + auth_data={**auth, 'reserved': value}).data, + rebuild, offset=16) + + def test_pcapng_zigbee_keys(self) -> None: + # The ZigBee NWK and APS Key Secrets Data open with a 16-octet key. + from pcapkit.const.pcapng.secrets_type import SecretsType + from pcapkit.protocols.misc.pcapng import PCAPNG + + proto = PCAPNG.__new__(PCAPNG) + for code, maker, arg, name in ( + (SecretsType.ZigBee_NWK_Key, proto._make_secrets_zigbee_nwk, 'nwk_key', '[ZigBee NWK Key] key'), + (SecretsType.ZigBee_APS_Key, proto._make_secrets_zigbee_aps, 'aps_key', '[ZigBee APS Key] key')): + with self.subTest(code=code): + schema = type(maker(code)) + self.assertWidth('PCAP-NG', name, 16, lambda value, maker=maker, code=code, arg=arg: + maker(code, **{arg: value}).pack(), + lambda data, schema=schema: schema.unpack(data).pack(), offset=0) + self.assertEqual(maker(code).pack()[:16], bytes(16)) + + +class CapturedOctetTests(_Case): + """A capture cut inside a reserved field rebuilds as captured; a model of the wrong width is refused. + + After a short read a padding field keeps only the octets present, where a + byte field zero-fills to its width. So a maker zero-fills a short value + from a model that records the short read (``__short_read__``), as the rest + of the cut header is, and the rebuild cuts back to what was captured + (#1458). Any other model's value is checked as a caller's is. + """ + + #: MH messages whose reserved octets no fixed field follows, so that no cut + #: past them records a short read: the options area after them records none. + NOTHING_AFTER = {'_make_msg_fna', '_make_msg_has', '_make_msg_sq'} + + #: MH messages whose cut inside the reserved octets does not parse, and why. + UNPARSED = { + '_make_msg_lri': 'a cut message reads its options area as zeros, which raises a FieldValueError', + '_make_msg_upa': 'a cut message reads its options area as zeros, which raises a FieldValueError', + } + + def _mh_messages(self) -> 'list[tuple[str, object, int]]': + from pcapkit.protocols.internet.mh import MH + + packets = {name: code for code, name in MH.__message__.items() if isinstance(name, str)} + rows = [] + for name in sorted(vars(MH)): + if name.startswith('_make_msg_'): + for arg, size in _octet_arguments(getattr(MH, name), set()): + if arg == 'msg_reserved': + rows.append((name, packets[name[len('_make_msg_'):]], size)) + return rows + + def _parse(self, cls: 'type', data: 'bytes', **kwargs: 'object') -> 'object': + """The info of ``data`` parsed by ``cls``, the warning of a cut length kept quiet.""" + from pcapkit.utilities.warnings import SchemaWarning + + with warnings.catch_warnings(): + warnings.simplefilter('ignore', SchemaWarning) + if kwargs: + return cls(io.BytesIO(data), len(data), **kwargs).info # type: ignore[call-arg] + return cls(data).info # type: ignore[call-arg] + + def assertCutRebuilds(self, cls: 'type', data: 'bytes', mark: 'bytes', **kwargs: 'object') -> None: + """Every cut of ``data`` inside ``mark``, or just before it, rebuilds as the octets captured. + + The rebuild is made from the data model and from the :obj:`dict` its + ``to_dict()`` gives, which ``from_data`` takes as well. + """ + start = data.index(mark) + for cut in range(max(start - 1, 1), start + len(mark)): + with self.subTest(protocol=cls.__name__, cut=cut): + part = data[:cut] + info = self._parse(cls, part, **kwargs) + self.assertIsNotNone(info.get('__short_read__')) # type: ignore[attr-defined] + self.assertEqual(cls.from_data(info).data, part) # type: ignore[attr-defined] + self.assertEqual(cls.from_data(info.to_dict()).data, part) # type: ignore[attr-defined] + + def assertEditRefused(self, cls: 'type', data: 'bytes', mark: 'bytes', edit: 'object', message: 'str', + **kwargs: 'object') -> 'bool': + """A capture cut past ``mark`` keeps it whole, so ``edit`` shortening it is refused with ``message``. + + The cut is the first past ``mark`` that records a short read; whether + there was one is returned. + """ + from pcapkit.utilities.exceptions import ProtocolError + + end = data.index(mark) + len(mark) + cuts = [cut for cut in range(end + 1, len(data)) + if self._parse(cls, data[:cut], **kwargs).get('__short_read__') is not None] # type: ignore[attr-defined] + if not cuts: + return False + for value in sorted({mark[:-1], b''}, key=len): + with self.subTest(protocol=cls.__name__, cut=cuts[0], width=len(value)): + info = self._parse(cls, data[:cuts[0]], **kwargs) + edit(info, value) # type: ignore[operator] + with self.assertRaises(ProtocolError) as caught: + cls.from_data(info) # type: ignore[attr-defined] + self.assertEqual(str(caught.exception), f'{message}, not {value!r}') + return True + + def test_mh_message_cut_inside_reserved_octets(self) -> None: + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import FieldValueError + + swept = [] + for name, code, size in self._mh_messages(): + swept.append(name) + mark = bytes(range(0xA1, 0xA1 + size)) + data = bytes(MH(next=59, type=code, data={'msg_reserved': mark})) + if name in self.UNPARSED: + # So the short read is recorded on a model parsed whole, as a + # cut there would record it: the field ``data`` follows the + # six octets of the MH header. + start = data.index(mark) + for count in range(size): + with self.subTest(maker=name, cut=start + count): + with self.assertRaises(FieldValueError): + MH(io.BytesIO(data[:start + count]), start + count, extension=True) + info = MH(io.BytesIO(data), len(data), extension=True).info + info.__update__({'msg_reserved': mark[:count], '__short_read__': ('data', start - 6 + count)}) + self.assertEqual(MH.from_data(info).data, data[:start + count]) + continue + with self.subTest(maker=name): + self.assertCutRebuilds(MH, data, mark, extension=True) + for name in ('_make_msg_fna', '_make_msg_has', '_make_msg_sq', *self.UNPARSED): + self.assertIn(name, swept) + + def test_ipv6_frag_and_route_cut_inside_reserved_octets(self) -> None: + # RFC 8200 section 4.5 and RFC 6275 section 6.4: the reserved octet + # follows Next Header; the reserved word follows Segments Left. + from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + self.assertCutRebuilds(IPv6_Frag, IPv6_Frag(next=59, reserved_octet=b'\xa1').data, b'\xa1', + extension=True) + for kind, args in ((0, {'ip': ['2001:db8::1']}), (2, {'ip': '2001:db8::1'})): + data = IPv6_Route(next=59, type=kind, data={**args, 'reserved': b'\xa1\xa2\xa3\xa4'}).data + with self.subTest(type=kind): + self.assertCutRebuilds(IPv6_Route, data, b'\xa1\xa2\xa3\xa4', extension=True) + + def test_ospf_cut_inside_authentication_reserved_octets(self) -> None: + # RFC 2328 appendix D.3: cryptographic authentication opens with two + # reserved octets, at octet 16 of the header. The dict form turns the + # authentication data model into a mapping, which is filled the same. + from pcapkit.const.ospf.authentication import Authentication + from pcapkit.protocols.application.ospf import OSPF + + data = OSPF(auth_type=Authentication.Cryptographic_authentication, + auth_data={'key_id': 1, 'len': 16, 'seq': 1, 'reserved': b'\xa1\xa2'}).data + self.assertEqual(data.index(b'\xa1\xa2'), 16) + start = data.index(b'\xa1\xa2') + for cut in range(start - 1, start + 2): + with self.subTest(cut=cut): + info = self._parse(OSPF, data[:cut]) + self.assertIsNotNone(info.get('__short_read__')) # type: ignore[attr-defined] + self.assertEqual(OSPF.from_data(info).data, data[:cut]) + self.assertEqual(OSPF.from_data(info.to_dict()).data, data[:cut]) # type: ignore[attr-defined] + + def test_cut_past_the_field_keeps_it_checked(self) -> None: + # A short read names where the capture ended; octets read whole before + # it are checked as a caller's are, an edit to them included. + from pcapkit.const.ospf.authentication import Authentication + from pcapkit.protocols.application.ospf import OSPF + from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + from pcapkit.protocols.internet.mh import MH + + def update(key: 'str') -> 'object': + return lambda info, value: info.__update__({key: value}) + + for name, code, size in self._mh_messages(): + if name in self.UNPARSED: + continue + mark = bytes(range(0xA1, 0xA1 + size)) + unit = 'octet' if size == 1 else 'octets' + with self.subTest(maker=name): + found = self.assertEditRefused(MH, bytes(MH(next=59, type=code, data={'msg_reserved': mark})), + mark, update('msg_reserved'), + f'MH: invalid reserved octets: must be {size} {unit}', + extension=True) + self.assertEqual(found, name not in self.NOTHING_AFTER) + mark = b'\xa1\xa2\xa3\xa4' + for kind, args in ((0, {'ip': ['2001:db8::1']}), (2, {'ip': '2001:db8::1'})): + with self.subTest(type=kind): + data = IPv6_Route(next=59, type=kind, data={**args, 'reserved': mark}).data + self.assertTrue(self.assertEditRefused(IPv6_Route, data, mark, update('reserved'), + 'IPv6-Route: invalid reserved octets: must be 4 octets', + extension=True)) + self.assertTrue(self.assertEditRefused(IPv6_Frag, IPv6_Frag(next=59, reserved_octet=b'\xa1').data, b'\xa1', + update('reserved_octet'), + 'IPv6-Frag: invalid reserved octet: must be 1 octet', extension=True)) + data = OSPF(auth_type=Authentication.Cryptographic_authentication, + auth_data={'key_id': 1, 'len': 16, 'seq': 1, 'reserved': b'\xa1\xa2'}).data + self.assertTrue(self.assertEditRefused(OSPF, data, b'\xa1\xa2', + lambda info, value: info.auth.__update__({'reserved': value}), + 'OSPF: invalid reserved octets: must be 2 octets')) + + def test_model_of_the_wrong_width_is_refused(self) -> None: + # A model parsed whole and edited records no short read, so its + # octets are checked as a caller's are, through from_data or given + # to the maker as ``message``. + from pcapkit.const.mh.packet import Packet + from pcapkit.const.ospf.authentication import Authentication + from pcapkit.protocols.application.ospf import OSPF + from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities.exceptions import ProtocolError + + frag = IPv6_Frag(next=59, reserved_octet=b'\xa1').data + for value in (b'', b'\xff\xff'): + with self.subTest(protocol='IPv6-Frag', width=len(value)): + info = IPv6_Frag(io.BytesIO(frag), len(frag), extension=True).info + info.__update__({'reserved_octet': value}) + with self.assertRaises(ProtocolError) as caught: + IPv6_Frag.from_data(info) + self.assertEqual(str(caught.exception), + f'IPv6-Frag: invalid reserved octet: must be 1 octet, not {value!r}') + hoti = bytes(MH(next=59, type=Packet.Home_Test_Init, data={'msg_reserved': b'\xa1\xa2'})) + for width in (0, 1, 3, 5): + value = b'\xff' * width + for form in ('from_data', 'message'): + with self.subTest(protocol='MH', width=width, form=form): + info = MH(io.BytesIO(hoti), len(hoti), extension=True).info + info.__update__({'msg_reserved': value}) + with self.assertRaises(ProtocolError) as caught: + if form == 'from_data': + MH.from_data(info) + else: + MH(next=59, type=Packet.Home_Test_Init, data={'message': info}) + self.assertEqual(str(caught.exception), + f'MH: invalid reserved octets: must be 2 octets, not {value!r}') + for kind, args in ((0, {'ip': ['2001:db8::1']}), (2, {'ip': '2001:db8::1'})): + route = IPv6_Route(next=59, type=kind, data={**args, 'reserved': b'\xa1\xa2\xa3\xa4'}).data + for width in (1, 6): + value = b'\xff' * width + with self.subTest(protocol='IPv6-Route', type=kind, width=width): + info = IPv6_Route(io.BytesIO(route), len(route), extension=True).info + info.__update__({'reserved': value}) + with self.assertRaises(ProtocolError) as caught: + IPv6_Route.from_data(info) + self.assertEqual(str(caught.exception), + f'IPv6-Route: invalid reserved octets: must be 4 octets, not {value!r}') + ospf = OSPF(auth_type=Authentication.Cryptographic_authentication, + auth_data={'key_id': 1, 'len': 16, 'seq': 1, 'reserved': b'\xa1\xa2'}).data + for width in (0, 1, 5): + value = b'\xff' * width + with self.subTest(protocol='OSPF', width=width): + info = OSPF(ospf).info + info.auth.__update__({'reserved': value}) + with self.assertRaises(ProtocolError) as caught: + OSPF.from_data(info) + self.assertEqual(str(caught.exception), + f'OSPF: invalid reserved octets: must be 2 octets, not {value!r}') + +if __name__ == '__main__': + unittest.main()