diff --git a/pcapkit/protocols/internet/esp.py b/pcapkit/protocols/internet/esp.py index 22e23ebddb..8598e3582d 100644 --- a/pcapkit/protocols/internet/esp.py +++ b/pcapkit/protocols/internet/esp.py @@ -542,78 +542,6 @@ class SecurityAssociation: """ - def __init__(self, spi: 'Optional[int]' = None, *, - encryption: 'Cipher | str | int' = Cipher.ENCR_NULL, - encryption_key: 'bytes' = b'', - salt: 'Optional[bytes]' = None, - integrity: 'Integrity | str | int' = Integrity.NONE, - integrity_key: 'bytes' = b'', - icv_length: 'Optional[int]' = None, - destination: 'Optional[IPv4Address | IPv6Address | str | int | bytes]' = None, - strict: 'bool' = True) -> 'None': - if spi is not None and not 0 <= spi <= 0xFFFFFFFF: - raise ProtocolError(f'ESP: invalid SPI: {spi!r}') - - if isinstance(destination, bool): - # NOTE: checked before the ``ipaddress.ip_address`` call below, for - # the same reason as ``_reject_bool`` in - # ``pcapkit.corekit.fields.ipaddress`` (this module calls - # ``ipaddress.ip_address`` directly rather than through a Field, so - # it needs its own copy of the guard): ``bool`` is an ``int`` - # subclass, and ``ipaddress.ip_address()`` treats any ``int`` below - # ``2**32`` as IPv4 -- so without this check, ``destination=True`` - # would silently become ``IPv4Address('0.0.0.1')``, with no - # exception and no warning. - raise ProtocolError( - f'ESP: invalid destination: must not be a bool, not {destination!r} -- ' - f'pass int({destination!r}) if the numeric value is what is wanted') - - #: Optional[int]: Security Parameters Index, or :data:`None` for any. - self.spi = spi - #: CipherSuite: How to apply the encryption algorithm. - self.cipher_suite = CipherSuite.get(encryption) - #: IntegritySuite: How to apply the integrity algorithm. - self.integrity_suite = IntegritySuite.get(integrity) - #: Cipher: Encryption algorithm, i.e. its IKEv2 transform. - self.encryption = self.cipher_suite.cipher - #: Integrity: Integrity algorithm, i.e. its IKEv2 transform. - self.integrity = self.integrity_suite.integrity - #: bool: Whether to enforce the :rfc:`4303` §2.4 padding pattern. - self.strict = strict - #: Optional[IPv4Address | IPv6Address]: Outer destination address. - self.destination = ipaddress.ip_address(destination) if destination is not None else None - - if self.cipher_suite.is_aead and self.integrity is not Integrity.NONE: - raise ProtocolError( - f'ESP: invalid integrity algorithm: {self.encryption.name} is a combined mode ' - f'algorithm and provides its own integrity, so {self.integrity.name!r} must ' - f'not be configured alongside it' - ) - - key, self.__salt__ = self._split_key(self.cipher_suite, encryption_key, salt) - self.__key__ = key - self.__integrity_key__ = bytes(integrity_key) - - if self.integrity is not Integrity.NONE: - expected = self.integrity_suite.key_size - if len(self.__integrity_key__) != expected: - # NOTE: #2411 -- the ``ESP:`` prefix matters more here than on - # the exceptions above: ``warn`` reports the caller's frame, so - # the message text is the only thing that names the protocol. - warn(f'ESP: invalid integrity key: {self.integrity.name} expects a ' - f'{expected}-octet key, got {len(self.__integrity_key__)} octets; ' - f'the ICV will very likely fail to verify', ProtocolWarning) - - if icv_length is not None and icv_length < 0: - raise ProtocolError(f'ESP: invalid ICV length: {icv_length!r}') - self.__icv_length__ = icv_length - - unavailable = self.unavailable() - if unavailable is not None: - warn(f'{unavailable}; ESP payloads for SPI ' - f'{"any" if self.spi is None else f"{self.spi:#010x}"} will be reported as ' - f'opaque ciphertext', ProtocolWarning) - ########################################################################## # Properties. ########################################################################## @@ -844,6 +772,89 @@ def encrypt(self, spi: 'int', seq: 'int', plaintext: 'bytes', ciphertext = encryptor.update(plaintext) + encryptor.finalize() return iv + ciphertext, encryptor.tag[:self.icv_length] + ########################################################################## + # Data models. + ########################################################################## + + def __init__(self, spi: 'Optional[int]' = None, *, + encryption: 'Cipher | str | int' = Cipher.ENCR_NULL, + encryption_key: 'bytes' = b'', + salt: 'Optional[bytes]' = None, + integrity: 'Integrity | str | int' = Integrity.NONE, + integrity_key: 'bytes' = b'', + icv_length: 'Optional[int]' = None, + destination: 'Optional[IPv4Address | IPv6Address | str | int | bytes]' = None, + strict: 'bool' = True) -> 'None': + if spi is not None and not 0 <= spi <= 0xFFFFFFFF: + raise ProtocolError(f'ESP: invalid SPI: {spi!r}') + + if isinstance(destination, bool): + # NOTE: checked before the ``ipaddress.ip_address`` call below, for + # the same reason as ``_reject_bool`` in + # ``pcapkit.corekit.fields.ipaddress`` (this module calls + # ``ipaddress.ip_address`` directly rather than through a Field, so + # it needs its own copy of the guard): ``bool`` is an ``int`` + # subclass, and ``ipaddress.ip_address()`` treats any ``int`` below + # ``2**32`` as IPv4 -- so without this check, ``destination=True`` + # would silently become ``IPv4Address('0.0.0.1')``, with no + # exception and no warning. + raise ProtocolError( + f'ESP: invalid destination: must not be a bool, not {destination!r} -- ' + f'pass int({destination!r}) if the numeric value is what is wanted') + + #: Optional[int]: Security Parameters Index, or :data:`None` for any. + self.spi = spi + #: CipherSuite: How to apply the encryption algorithm. + self.cipher_suite = CipherSuite.get(encryption) + #: IntegritySuite: How to apply the integrity algorithm. + self.integrity_suite = IntegritySuite.get(integrity) + #: Cipher: Encryption algorithm, i.e. its IKEv2 transform. + self.encryption = self.cipher_suite.cipher + #: Integrity: Integrity algorithm, i.e. its IKEv2 transform. + self.integrity = self.integrity_suite.integrity + #: bool: Whether to enforce the :rfc:`4303` §2.4 padding pattern. + self.strict = strict + #: Optional[IPv4Address | IPv6Address]: Outer destination address. + self.destination = ipaddress.ip_address(destination) if destination is not None else None + + if self.cipher_suite.is_aead and self.integrity is not Integrity.NONE: + raise ProtocolError( + f'ESP: invalid integrity algorithm: {self.encryption.name} is a combined mode ' + f'algorithm and provides its own integrity, so {self.integrity.name!r} must ' + f'not be configured alongside it' + ) + + key, self.__salt__ = self._split_key(self.cipher_suite, encryption_key, salt) + self.__key__ = key + self.__integrity_key__ = bytes(integrity_key) + + if self.integrity is not Integrity.NONE: + expected = self.integrity_suite.key_size + if len(self.__integrity_key__) != expected: + # NOTE: #2411 -- the ``ESP:`` prefix matters more here than on + # the exceptions above: ``warn`` reports the caller's frame, so + # the message text is the only thing that names the protocol. + warn(f'ESP: invalid integrity key: {self.integrity.name} expects a ' + f'{expected}-octet key, got {len(self.__integrity_key__)} octets; ' + f'the ICV will very likely fail to verify', ProtocolWarning) + + if icv_length is not None and icv_length < 0: + raise ProtocolError(f'ESP: invalid ICV length: {icv_length!r}') + self.__icv_length__ = icv_length + + unavailable = self.unavailable() + if unavailable is not None: + warn(f'{unavailable}; ESP payloads for SPI ' + f'{"any" if self.spi is None else f"{self.spi:#010x}"} will be reported as ' + f'opaque ciphertext', ProtocolWarning) + + def __repr__(self) -> 'str': + """Representation of the SA, free of any key material.""" + spi = 'any' if self.spi is None else f'{self.spi:#010x}' + dst = '' if self.destination is None else f' dst={self.destination!s}' + return (f'') + ########################################################################## # Utilities. ########################################################################## @@ -887,17 +898,6 @@ def _split_key(suite: 'CipherSuite', material: 'bytes', f'got {len(material)!r}') return material, salt - ########################################################################## - # Data models. - ########################################################################## - - def __repr__(self) -> 'str': - """Representation of the SA, free of any key material.""" - spi = 'any' if self.spi is None else f'{self.spi:#010x}' - dst = '' if self.destination is None else f' dst={self.destination!s}' - return (f'') - class ESPContext(ProtocolContext): """Caller supplied Security Association context for :class:`ESP`. @@ -908,11 +908,6 @@ class ESPContext(ProtocolContext): """ - def __init__(self, *associations: 'SecurityAssociation') -> 'None': - self.__associations__ = [] # type: list[SecurityAssociation] - for association in associations: - self.register(association) - ########################################################################## # Properties. ########################################################################## @@ -970,6 +965,11 @@ def match(self, spi: 'int', # Data models. ########################################################################## + def __init__(self, *associations: 'SecurityAssociation') -> 'None': + self.__associations__ = [] # type: list[SecurityAssociation] + for association in associations: + self.register(association) + def __repr__(self) -> 'str': """Representation of the context, free of any key material.""" return f'' @@ -1393,32 +1393,55 @@ def _make_data(cls, data: 'Data_ESP') -> 'dict[str, Any]': # type: ignore[overr 'payload': data.payload_data + data.icv, } - @staticmethod - def _payload_bytes(payload: 'bytes | ProtocolBase | Schema') -> 'bytes': - """Render ``payload`` as :obj:`bytes`. + def _make_opaque(self, spi: 'int', seq: 'int', total: 'int', payload_data: 'bytes', + status: 'ESPStatus', error: 'str', *, icv: 'bytes' = b'', + version: 'Literal[4, 6]' = 4, + packet: 'Optional[dict[str, Any]]' = None, + warning: 'bool' = True) -> 'Data_ESP': + """Report an ESP packet whose payload was not decrypted. + + The payload is surfaced as :class:`~pcapkit.protocols.misc.raw.Raw` + -- which is what a next header of :data:`None` resolves to -- so that + :attr:`self.payload ` + and the protocol chain behave as they do for any other protocol, and + the trailer fields are left :data:`None` rather than guessed at. Args: - payload: Payload as supplied to :meth:`make`. + spi: Security Parameters Index. + seq: Sequence number. + total: Total length of the ESP portion of the packet. + payload_data: Payload data, excluding the ICV. + status: Why the payload was not decrypted. + error: Human readable form of ``status``. + icv: Integrity check value, when its length is known. + version: IP protocol version. + packet: Packet information from the enclosing layer. + warning: Whether to warn; a capture taken without keys is the + expected case and does not warrant one. Returns: - Packed payload. - - Raises: - ProtocolUnbound: If ``payload`` is of an unsupported type. This - mirrors :meth:`Schema.pack `, - which rejects the same set. + Parsed packet data. """ - from pcapkit.protocols.protocol import \ - ProtocolBase # pylint: disable=import-outside-toplevel + if warning: + warn(error, ProtocolWarning) - if isinstance(payload, bytes): - return payload - if isinstance(payload, Schema): - return payload.pack() - if isinstance(payload, ProtocolBase): - return bytes(payload) - raise ProtocolUnbound(f'unsupported type {type(payload)}') + esp = Data_ESP( + spi=spi, + seq=seq, + length=total, + payload_data=payload_data, + icv=icv, + status=status, + error=error, + next=None, + pad_len=None, + padding=None, + plaintext=None, + ) + return self._decode_next_layer(esp, None, len(payload_data) + len(icv), + packet=packet or None, version=version, + payload=payload_data + icv) @staticmethod def _read_trailer(plaintext: 'bytes', association: 'SecurityAssociation', @@ -1461,52 +1484,29 @@ def _read_trailer(plaintext: 'bytes', association: 'SecurityAssociation', return inner, padding, pad_len, next_ - def _make_opaque(self, spi: 'int', seq: 'int', total: 'int', payload_data: 'bytes', - status: 'ESPStatus', error: 'str', *, icv: 'bytes' = b'', - version: 'Literal[4, 6]' = 4, - packet: 'Optional[dict[str, Any]]' = None, - warning: 'bool' = True) -> 'Data_ESP': - """Report an ESP packet whose payload was not decrypted. - - The payload is surfaced as :class:`~pcapkit.protocols.misc.raw.Raw` - -- which is what a next header of :data:`None` resolves to -- so that - :attr:`self.payload ` - and the protocol chain behave as they do for any other protocol, and - the trailer fields are left :data:`None` rather than guessed at. + @staticmethod + def _payload_bytes(payload: 'bytes | ProtocolBase | Schema') -> 'bytes': + """Render ``payload`` as :obj:`bytes`. Args: - spi: Security Parameters Index. - seq: Sequence number. - total: Total length of the ESP portion of the packet. - payload_data: Payload data, excluding the ICV. - status: Why the payload was not decrypted. - error: Human readable form of ``status``. - icv: Integrity check value, when its length is known. - version: IP protocol version. - packet: Packet information from the enclosing layer. - warning: Whether to warn; a capture taken without keys is the - expected case and does not warrant one. + payload: Payload as supplied to :meth:`make`. Returns: - Parsed packet data. + Packed payload. + + Raises: + ProtocolUnbound: If ``payload`` is of an unsupported type. This + mirrors :meth:`Schema.pack `, + which rejects the same set. """ - if warning: - warn(error, ProtocolWarning) + from pcapkit.protocols.protocol import \ + ProtocolBase # pylint: disable=import-outside-toplevel - esp = Data_ESP( - spi=spi, - seq=seq, - length=total, - payload_data=payload_data, - icv=icv, - status=status, - error=error, - next=None, - pad_len=None, - padding=None, - plaintext=None, - ) - return self._decode_next_layer(esp, None, len(payload_data) + len(icv), - packet=packet or None, version=version, - payload=payload_data + icv) + if isinstance(payload, bytes): + return payload + if isinstance(payload, Schema): + return payload.pack() + if isinstance(payload, ProtocolBase): + return bytes(payload) + raise ProtocolUnbound(f'unsupported type {type(payload)}') diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index 08c76454ad..a7938c9f92 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -886,6 +886,80 @@ def _read_hip_param(self, length: 'int', *, version: 'int') -> 'Parameter': # p raise ProtocolError(f'HIPv{version}: invalid format') return options + def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parameter, Data_Parameter | dict[str, Any]] | bytes] | Parameter', *, # pylint: disable=line-too-long + version: 'int') -> 'tuple[list[Schema_Parameter | bytes], int]': + """Make HIP parameter. + + A ``(code, value)`` pair carries either a mapping of keyword arguments + for the parameter's constructor, or a parsed + :class:`~pcapkit.protocols.data.internet.hip.Parameter` data model, + which goes to the constructor positionally through + :meth:`_make_hip_param_data` rather than expanded as keyword arguments + (:issue:`2352`). + + Args: + parameters: HIP parameters + version: HIP protocol version + + Returns: + HIP parameters and total length. + + """ + total_length = 0 + if isinstance(parameters, list): + parameters_list = [] # type: list[Schema_Parameter | bytes] + for schema in parameters: + if isinstance(schema, bytes): + # :rfc:`7401#section-5.2.1` pads every parameter to a + # multiple of 8 octets, and Header Length counts in those + # units; a raw parameter is taken to omit its padding. + schema += bytes(-len(schema) % 8) + parameters_list.append(schema) + total_length += len(schema) + elif isinstance(schema, Schema): + schema_packed = schema.pack() + + parameters_list.append(schema_packed) + total_length += len(schema_packed) + else: + code, args = cast('tuple[Enum_Parameter, Data_Parameter | dict[str, Any]]', + schema) + name = self._lookup_registry(self.__parameter__, code) + if isinstance(name, str): + meth_name = f'_make_param_{name}' + meth = cast('ParameterConstructor', + getattr(self, meth_name, self._make_param_unassigned)) + else: + meth = name[1] + + if isinstance(args, Data_Parameter): + data = self._make_hip_param_data(meth, code, args, + version=version) # type: Schema_Parameter + else: + data = meth(code, version=version, **args) + data_packed = data.pack() + + parameters_list.append(data) + total_length += len(data_packed) + return parameters_list, total_length + + parameters_list = [] + for code, param in parameters.items(multi=True): + name = self._lookup_registry(self.__parameter__, code) + if isinstance(name, str): + meth_name = f'_make_param_{name}' + meth = cast('ParameterConstructor', + getattr(self, meth_name, self._make_param_unassigned)) + else: + meth = name[1] + + data = self._make_hip_param_data(meth, code, param, version=version) + data_packed = data.pack() + + parameters_list.append(data) + total_length += len(data_packed) + return parameters_list, total_length + def _read_param_unassigned(self, schema: 'Schema_UnassignedParameter', *, version: 'int', # pylint: disable=unused-argument options: 'Parameter') -> 'Data_UnassignedParameter': # pylint: disable=unused-argument """Read HIP unassigned parameters. @@ -3199,80 +3273,6 @@ def _read_param_relay_hmac(self, schema: 'Schema_RelayHMACParameter', *, version ) return relay_hmac - def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parameter, Data_Parameter | dict[str, Any]] | bytes] | Parameter', *, # pylint: disable=line-too-long - version: 'int') -> 'tuple[list[Schema_Parameter | bytes], int]': - """Make HIP parameter. - - A ``(code, value)`` pair carries either a mapping of keyword arguments - for the parameter's constructor, or a parsed - :class:`~pcapkit.protocols.data.internet.hip.Parameter` data model, - which goes to the constructor positionally through - :meth:`_make_hip_param_data` rather than expanded as keyword arguments - (:issue:`2352`). - - Args: - parameters: HIP parameters - version: HIP protocol version - - Returns: - HIP parameters and total length. - - """ - total_length = 0 - if isinstance(parameters, list): - parameters_list = [] # type: list[Schema_Parameter | bytes] - for schema in parameters: - if isinstance(schema, bytes): - # :rfc:`7401#section-5.2.1` pads every parameter to a - # multiple of 8 octets, and Header Length counts in those - # units; a raw parameter is taken to omit its padding. - schema += bytes(-len(schema) % 8) - parameters_list.append(schema) - total_length += len(schema) - elif isinstance(schema, Schema): - schema_packed = schema.pack() - - parameters_list.append(schema_packed) - total_length += len(schema_packed) - else: - code, args = cast('tuple[Enum_Parameter, Data_Parameter | dict[str, Any]]', - schema) - name = self._lookup_registry(self.__parameter__, code) - if isinstance(name, str): - meth_name = f'_make_param_{name}' - meth = cast('ParameterConstructor', - getattr(self, meth_name, self._make_param_unassigned)) - else: - meth = name[1] - - if isinstance(args, Data_Parameter): - data = self._make_hip_param_data(meth, code, args, - version=version) # type: Schema_Parameter - else: - data = meth(code, version=version, **args) - data_packed = data.pack() - - parameters_list.append(data) - total_length += len(data_packed) - return parameters_list, total_length - - parameters_list = [] - for code, param in parameters.items(multi=True): - name = self._lookup_registry(self.__parameter__, code) - if isinstance(name, str): - meth_name = f'_make_param_{name}' - meth = cast('ParameterConstructor', - getattr(self, meth_name, self._make_param_unassigned)) - else: - meth = name[1] - - data = self._make_hip_param_data(meth, code, param, version=version) - data_packed = data.pack() - - parameters_list.append(data) - total_length += len(data_packed) - return parameters_list, total_length - def _make_hip_param_data(self, meth: 'ParameterConstructor', code: 'Enum_Parameter', param: 'Data_Parameter', *, version: 'int') -> 'Schema_Parameter': diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index c72adf2555..eca9e0dbf6 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -596,6 +596,104 @@ def _read_hopopt_options(self, length: 'int') -> 'Option': raise ProtocolError('HOPOPT: invalid format') return options + def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option') -> 'tuple[list[Schema_Option | bytes], int]': + """Make options for HOPOPT. + + Args: + options: HOPOPT options + + Returns: + Tuple of options and total length of options. + + Note: + Options given as a :obj:`list` are fresh input: padding options in + it are dropped, and padding is inserted after each option to align + the header. Options given as a mapping, such as the + :class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` of parsed + data, are emitted in order with their own padding options kept, and + padding is added only at the end if the header is not aligned. + + The returned length is that of the options area alone, the value + :meth:`make` derives the ``Hdr Ext Len`` field from. Since the two + octets of the fixed header precede it, an options area is well + formed only when it is 2 octets short of a multiple of 8 -- + including when it holds no real options at all, in which case it is + 6 octets of pure padding rather than empty. + + """ + total_length = 0 + if isinstance(options, list): + options_list = [] # type: list[Schema_Option | bytes] + for schema in options: + if isinstance(schema, bytes): + code = Enum_Option.get(schema[0]) + if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default + continue + + opt = schema # type: bytes | Schema_Option + opt_len = len(schema) + elif isinstance(schema, Schema): + code = schema.type + if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default + continue + + opt = schema + opt_len = len(schema.pack()) + else: + code, args = cast('tuple[Enum_Option, dict[str, Any]]', schema) + if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default + continue + + name = self._lookup_registry(self.__option__, code) + if isinstance(name, str): + meth_name = f'_make_opt_{name}' + meth = cast('OptionConstructor', + getattr(self, meth_name, self._make_opt_none)) + else: + meth = name[1] + + opt = meth(code, **args) + opt_len = len(opt.pack()) + + options_list.append(opt) + total_length += opt_len + + # force alignment of the header (fixed part included) to 8 octets + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len + + # an options area holding nothing at all is still 6 octets long + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len + return options_list, total_length + + # NOTE: Options given as parsed data are emitted as they are, padding + # included and in place, so that a parsed header rebuilds byte for + # byte; only a header left short of alignment gets trailing padding. + options_list = [] + for code, option in options.items(multi=True): + name = self._lookup_registry(self.__option__, code) + if isinstance(name, str): + meth_name = f'_make_opt_{name}' + meth = cast('OptionConstructor', + getattr(self, meth_name, self._make_opt_none)) + else: + meth = name[1] + + opt = meth(code, option) + opt_len = len(opt.pack()) + + options_list.append(opt) + total_length += opt_len + + # an options area holding nothing at all is still 6 octets long + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len + return options_list, total_length + def _read_opt_none(self, schema: 'Schema_UnassignedOption', *, options: 'Option') -> 'Data_UnassignedOption': # pylint: disable=unused-argument """Read HOPOPT unassigned options. @@ -1389,104 +1487,6 @@ def _make_pad_options(self, offset: 'int') -> 'tuple[list[Schema_PadOption], int return [self._make_opt_pad(Enum_Option.Pad1, length=0)], pad_len # type: ignore[arg-type] return [self._make_opt_pad(Enum_Option.PadN, length=pad_len - 2)], pad_len # type: ignore[arg-type] - def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option') -> 'tuple[list[Schema_Option | bytes], int]': - """Make options for HOPOPT. - - Args: - options: HOPOPT options - - Returns: - Tuple of options and total length of options. - - Note: - Options given as a :obj:`list` are fresh input: padding options in - it are dropped, and padding is inserted after each option to align - the header. Options given as a mapping, such as the - :class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` of parsed - data, are emitted in order with their own padding options kept, and - padding is added only at the end if the header is not aligned. - - The returned length is that of the options area alone, the value - :meth:`make` derives the ``Hdr Ext Len`` field from. Since the two - octets of the fixed header precede it, an options area is well - formed only when it is 2 octets short of a multiple of 8 -- - including when it holds no real options at all, in which case it is - 6 octets of pure padding rather than empty. - - """ - total_length = 0 - if isinstance(options, list): - options_list = [] # type: list[Schema_Option | bytes] - for schema in options: - if isinstance(schema, bytes): - code = Enum_Option.get(schema[0]) - if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default - continue - - opt = schema # type: bytes | Schema_Option - opt_len = len(schema) - elif isinstance(schema, Schema): - code = schema.type - if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default - continue - - opt = schema - opt_len = len(schema.pack()) - else: - code, args = cast('tuple[Enum_Option, dict[str, Any]]', schema) - if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default - continue - - name = self._lookup_registry(self.__option__, code) - if isinstance(name, str): - meth_name = f'_make_opt_{name}' - meth = cast('OptionConstructor', - getattr(self, meth_name, self._make_opt_none)) - else: - meth = name[1] - - opt = meth(code, **args) - opt_len = len(opt.pack()) - - options_list.append(opt) - total_length += opt_len - - # force alignment of the header (fixed part included) to 8 octets - pad_opts, pad_len = self._make_pad_options(total_length + 2) - options_list.extend(pad_opts) - total_length += pad_len - - # an options area holding nothing at all is still 6 octets long - pad_opts, pad_len = self._make_pad_options(total_length + 2) - options_list.extend(pad_opts) - total_length += pad_len - return options_list, total_length - - # NOTE: Options given as parsed data are emitted as they are, padding - # included and in place, so that a parsed header rebuilds byte for - # byte; only a header left short of alignment gets trailing padding. - options_list = [] - for code, option in options.items(multi=True): - name = self._lookup_registry(self.__option__, code) - if isinstance(name, str): - meth_name = f'_make_opt_{name}' - meth = cast('OptionConstructor', - getattr(self, meth_name, self._make_opt_none)) - else: - meth = name[1] - - opt = meth(code, option) - opt_len = len(opt.pack()) - - options_list.append(opt) - total_length += opt_len - - # an options area holding nothing at all is still 6 octets long - pad_opts, pad_len = self._make_pad_options(total_length + 2) - options_list.extend(pad_opts) - total_length += pad_len - return options_list, total_length - def _make_opt_none(self, code: 'Enum_Option', opt: 'Optional[Data_UnassignedOption]' = None, *, data: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_UnassignedOption': diff --git a/pcapkit/protocols/internet/ipsec.py b/pcapkit/protocols/internet/ipsec.py index 1b01658db4..aa80d77148 100644 --- a/pcapkit/protocols/internet/ipsec.py +++ b/pcapkit/protocols/internet/ipsec.py @@ -34,7 +34,7 @@ class IPsec(Internet[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract- """ ########################################################################## - # Data models. + # Methods. ########################################################################## @classmethod diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index 32181ad03e..75d6628ebb 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -584,6 +584,104 @@ def _read_ipv6_opts(self, length: 'int') -> 'Option': raise ProtocolError('IPv6-Opts: invalid format') return options + def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option') -> 'tuple[list[Schema_Option | bytes], int]': + """Make options for IPv6-Opts. + + Args: + options: IPv6-Opts options + + Returns: + Tuple of options and total length of options. + + Note: + Options given as a :obj:`list` are fresh input: padding options in + it are dropped, and padding is inserted after each option to align + the header. Options given as a mapping, such as the + :class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` of parsed + data, are emitted in order with their own padding options kept, and + padding is added only at the end if the header is not aligned. + + The returned length is that of the options area alone, the value + :meth:`make` derives the ``Hdr Ext Len`` field from. Since the two + octets of the fixed header precede it, an options area is well + formed only when it is 2 octets short of a multiple of 8 -- + including when it holds no real options at all, in which case it is + 6 octets of pure padding rather than empty. + + """ + total_length = 0 + if isinstance(options, list): + options_list = [] # type: list[Schema_Option | bytes] + for schema in options: + if isinstance(schema, bytes): + code = Enum_Option.get(schema[0]) + if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default + continue + + opt = schema # type: bytes | Schema_Option + opt_len = len(schema) + elif isinstance(schema, Schema): + code = schema.type + if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default + continue + + opt = schema + opt_len = len(schema.pack()) + else: + code, args = cast('tuple[Enum_Option, dict[str, Any]]', schema) + if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default + continue + + name = self._lookup_registry(self.__option__, code) + if isinstance(name, str): + meth_name = f'_make_opt_{name}' + meth = cast('OptionConstructor', + getattr(self, meth_name, self._make_opt_none)) + else: + meth = name[1] + + opt = meth(code, **args) + opt_len = len(opt.pack()) + + options_list.append(opt) + total_length += opt_len + + # force alignment of the header (fixed part included) to 8 octets + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len + + # an options area holding nothing at all is still 6 octets long + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len + return options_list, total_length + + # NOTE: Options given as parsed data are emitted as they are, padding + # included and in place, so that a parsed header rebuilds byte for + # byte; only a header left short of alignment gets trailing padding. + options_list = [] + for code, option in options.items(multi=True): + name = self._lookup_registry(self.__option__, code) + if isinstance(name, str): + meth_name = f'_make_opt_{name}' + meth = cast('OptionConstructor', + getattr(self, meth_name, self._make_opt_none)) + else: + meth = name[1] + + opt = meth(code, option) + opt_len = len(opt.pack()) + + options_list.append(opt) + total_length += opt_len + + # an options area holding nothing at all is still 6 octets long + pad_opts, pad_len = self._make_pad_options(total_length + 2) + options_list.extend(pad_opts) + total_length += pad_len + return options_list, total_length + def _read_opt_none(self, schema: 'Schema_UnassignedOption', *, options: 'Option') -> 'Data_UnassignedOption': # pylint: disable=unused-argument """Read IPv6-Opts unassigned options. @@ -1377,104 +1475,6 @@ def _make_pad_options(self, offset: 'int') -> 'tuple[list[Schema_PadOption], int return [self._make_opt_pad(Enum_Option.Pad1, length=0)], pad_len # type: ignore[arg-type] return [self._make_opt_pad(Enum_Option.PadN, length=pad_len - 2)], pad_len # type: ignore[arg-type] - def _make_ipv6_opts(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option') -> 'tuple[list[Schema_Option | bytes], int]': - """Make options for IPv6-Opts. - - Args: - options: IPv6-Opts options - - Returns: - Tuple of options and total length of options. - - Note: - Options given as a :obj:`list` are fresh input: padding options in - it are dropped, and padding is inserted after each option to align - the header. Options given as a mapping, such as the - :class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` of parsed - data, are emitted in order with their own padding options kept, and - padding is added only at the end if the header is not aligned. - - The returned length is that of the options area alone, the value - :meth:`make` derives the ``Hdr Ext Len`` field from. Since the two - octets of the fixed header precede it, an options area is well - formed only when it is 2 octets short of a multiple of 8 -- - including when it holds no real options at all, in which case it is - 6 octets of pure padding rather than empty. - - """ - total_length = 0 - if isinstance(options, list): - options_list = [] # type: list[Schema_Option | bytes] - for schema in options: - if isinstance(schema, bytes): - code = Enum_Option.get(schema[0]) - if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default - continue - - opt = schema # type: bytes | Schema_Option - opt_len = len(schema) - elif isinstance(schema, Schema): - code = schema.type - if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default - continue - - opt = schema - opt_len = len(schema.pack()) - else: - code, args = cast('tuple[Enum_Option, dict[str, Any]]', schema) - if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default - continue - - name = self._lookup_registry(self.__option__, code) - if isinstance(name, str): - meth_name = f'_make_opt_{name}' - meth = cast('OptionConstructor', - getattr(self, meth_name, self._make_opt_none)) - else: - meth = name[1] - - opt = meth(code, **args) - opt_len = len(opt.pack()) - - options_list.append(opt) - total_length += opt_len - - # force alignment of the header (fixed part included) to 8 octets - pad_opts, pad_len = self._make_pad_options(total_length + 2) - options_list.extend(pad_opts) - total_length += pad_len - - # an options area holding nothing at all is still 6 octets long - pad_opts, pad_len = self._make_pad_options(total_length + 2) - options_list.extend(pad_opts) - total_length += pad_len - return options_list, total_length - - # NOTE: Options given as parsed data are emitted as they are, padding - # included and in place, so that a parsed header rebuilds byte for - # byte; only a header left short of alignment gets trailing padding. - options_list = [] - for code, option in options.items(multi=True): - name = self._lookup_registry(self.__option__, code) - if isinstance(name, str): - meth_name = f'_make_opt_{name}' - meth = cast('OptionConstructor', - getattr(self, meth_name, self._make_opt_none)) - else: - meth = name[1] - - opt = meth(code, option) - opt_len = len(opt.pack()) - - options_list.append(opt) - total_length += opt_len - - # an options area holding nothing at all is still 6 octets long - pad_opts, pad_len = self._make_pad_options(total_length + 2) - options_list.extend(pad_opts) - total_length += pad_len - return options_list, total_length - def _make_opt_none(self, code: 'Enum_Option', opt: 'Optional[Data_UnassignedOption]' = None, *, data: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_UnassignedOption': diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index 87199b1e59..311fe527ed 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/pcapkit/protocols/internet/ipv6_route.py @@ -241,36 +241,6 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, # return ipv6_route return self._decode_next_layer(ipv6_route, schema.next, length - ipv6_route.length) - @staticmethod - def _make_hdr_ext_len(data_length: 'int') -> 'int': - """Compute ``Hdr Ext Len`` for a type-specific data payload of ``data_length`` octets. - - Per :rfc:`8200#section-4.4`, ``Hdr Ext Len`` is the length of the - Routing header in 8-octet units, not including the first 8 octets. The - fixed part (``next``/``length``/``type``/``seg_left``) is 4 octets, so - the on-the-wire header is ``4 + data_length`` octets; equating that to - ``8 + 8 * hdr_ext_len`` gives ``hdr_ext_len = (data_length - 4) / 8``. - :func:`~pcapkit.protocols.schema.internet.ipv6_route.ipv6_route_data_length` - is the inverse, used on the read side. - - This is the *only* place ``Hdr Ext Len`` is computed on the write side: - separate per-branch copies in :meth:`make` drift apart, so keep one - helper. Do NOT "simplify" the ``- 4`` / ``/ 8`` - away: the units either side differ (octets vs. 8-octet units), and - dropping the offset silently reinterprets the field. - - Args: - data_length: packed length, in octets, of the type-specific data - (i.e. ``len(data_val.pack())`` for a - :class:`~pcapkit.protocols.schema.schema.Schema`-based payload, or the padded raw - :obj:`bytes` length for the unknown/raw-bytes case). - - Returns: - The ``Hdr Ext Len`` field value. - - """ - return math.ceil(max(0, data_length - 4) / 8) - def make(self, dst: 'Optional[IPv6Address | str | int| bytes]' = None, next: 'Enum_TransType | StdlibEnum | AenumEnum | str | int' = Enum_TransType.UDP, @@ -514,6 +484,36 @@ def _make_data(cls, data: 'Data_IPv6_Route') -> 'dict[str, Any]': # type: ignor 'payload': cls._make_payload(data), } + @staticmethod + def _make_hdr_ext_len(data_length: 'int') -> 'int': + """Compute ``Hdr Ext Len`` for a type-specific data payload of ``data_length`` octets. + + Per :rfc:`8200#section-4.4`, ``Hdr Ext Len`` is the length of the + Routing header in 8-octet units, not including the first 8 octets. The + fixed part (``next``/``length``/``type``/``seg_left``) is 4 octets, so + the on-the-wire header is ``4 + data_length`` octets; equating that to + ``8 + 8 * hdr_ext_len`` gives ``hdr_ext_len = (data_length - 4) / 8``. + :func:`~pcapkit.protocols.schema.internet.ipv6_route.ipv6_route_data_length` + is the inverse, used on the read side. + + This is the *only* place ``Hdr Ext Len`` is computed on the write side: + separate per-branch copies in :meth:`make` drift apart, so keep one + helper. Do NOT "simplify" the ``- 4`` / ``/ 8`` + away: the units either side differ (octets vs. 8-octet units), and + dropping the offset silently reinterprets the field. + + Args: + data_length: packed length, in octets, of the type-specific data + (i.e. ``len(data_val.pack())`` for a + :class:`~pcapkit.protocols.schema.schema.Schema`-based payload, or the padded raw + :obj:`bytes` length for the unknown/raw-bytes case). + + Returns: + The ``Hdr Ext Len`` field value. + + """ + return math.ceil(max(0, data_length - 4) / 8) + def _read_data_type_none(self, schema: 'Schema_UnknownType', *, header: 'Schema_IPv6_Route') -> 'Data_UnknownType': """Read IPv6-Route unknown type data. diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 37edc5f158..2f8d812817 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -1396,30 +1396,6 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4, return mh return self._decode_next_layer(mh, schema.next, length - mh.length) - @staticmethod - def _mh_message_length(header_len: 'int') -> 'int': - """Compute the total on-the-wire MH header length for a given ``Header Len``. - - Per :rfc:`6275#section-6.1.1`, ``Header Len`` is *"the length of the - Mobility Header, in units of 8 octets, excluding the first 8 - octets"* -- i.e. the total header is ``8 + 8 * header_len`` octets, - or equivalently ``(header_len + 1) * 8``. Every ``_read_msg_*`` - below reports this value as the parsed message's own ``.length``, - which :meth:`~pcapkit.protocols.internet.mh.MH.read` subtracts from the - outer packet length to find the next layer's length. It is the inverse - of :meth:`make`'s ``(len(data_val) + 6) // 8 - 1``. Do NOT drop the - ``+ 1``: the units either side of it differ (octets vs. 8-octet - units), and dropping the offset silently reinterprets the field. - - Args: - header_len: raw ``Header Len`` field value, as read off the wire. - - Returns: - Total length, in octets, of the on-the-wire MH header. - - """ - return (header_len + 1) * 8 - def make(self, next: 'Enum_TransType | StdlibEnum | AenumEnum | str | int' = Enum_TransType.UDP, next_default: 'Optional[int]' = None, @@ -1670,6 +1646,94 @@ def _make_data(cls, data: 'Data_MH') -> 'dict[str, Any]': # type: ignore[overri 'payload': cls._make_payload(data), } + @staticmethod + def _mh_message_length(header_len: 'int') -> 'int': + """Compute the total on-the-wire MH header length for a given ``Header Len``. + + Per :rfc:`6275#section-6.1.1`, ``Header Len`` is *"the length of the + Mobility Header, in units of 8 octets, excluding the first 8 + octets"* -- i.e. the total header is ``8 + 8 * header_len`` octets, + or equivalently ``(header_len + 1) * 8``. Every ``_read_msg_*`` + below reports this value as the parsed message's own ``.length``, + which :meth:`~pcapkit.protocols.internet.mh.MH.read` subtracts from the + outer packet length to find the next layer's length. It is the inverse + of :meth:`make`'s ``(len(data_val) + 6) // 8 - 1``. Do NOT drop the + ``+ 1``: the units either side of it differ (octets vs. 8-octet + units), and dropping the offset silently reinterprets the field. + + Args: + header_len: raw ``Header Len`` field value, as read off the wire. + + Returns: + Total length, in octets, of the on-the-wire MH header. + + """ + return (header_len + 1) * 8 + + def _pad_mh_message(self, data: 'Schema_Packet | bytes') -> 'Schema_Packet | bytes': + """Pad an MH message so that the Mobility Header aligns to 8 octets. + + Args: + data: Constructed message data. + + Returns: + The message data, with padding options appended if any were needed. + + Note: + The padding goes into the message's mobility options, which is where + :rfc:`6275#section-6.1.1` puts it. A message whose body is opaque has + nowhere to put it -- + :class:`~pcapkit.protocols.schema.internet.mh.UnknownMessage` and + :class:`~pcapkit.protocols.schema.internet.mh.ExperimentalMessage` + carry raw bytes and no options, as does a ``data`` argument given + directly as :obj:`bytes` -- so for those the padding is appended to the + message body itself and a + :class:`~pcapkit.utilities.warnings.ProtocolWarning` says so. + + Appending is necessary rather than optional: ``length`` is + ``(len(data) + 6) // 8 - 1``, which floors, so leaving an opaque body + short would emit 10, 12 or 14 octets while declaring 8, and a parser reads + 8 and misinterprets the remainder. Since the caller asked for a packet + to be built and the shortfall is recoverable, completing it beats + refusing -- the warning is there because the emitted body is then not + byte-for-byte what was handed in. + + """ + pad_opts, pad_len = self._make_pad_options(len(data) + 6) + if pad_len == 0: + return data + + options = getattr(data, 'options', None) + # NOTE: The ``isinstance`` test comes first so that the type checker can + # narrow ``data`` for the assignment below; at runtime ``getattr`` has + # already covered the :obj:`bytes` case by returning :obj:`None`. + if isinstance(data, bytes) or options is None: + warn(f'{self.alias}: message data of {len(data)} octets carries no ' + f'mobility options to hold padding, so {pad_len} octet(s) were ' + 'appended to the message body to align the header', + ProtocolWarning) + if isinstance(data, bytes): + return data + b'\x00' * pad_len + + # An opaque schema body -- UnknownMessage, ExperimentalMessage -- keeps + # its content in ``data`` rather than in options, so that is where the + # octets go. Rebound rather than mutated in place so that ``len()`` and + # ``pack()`` see the change, exactly as for the options branch below. + body = getattr(data, 'data', None) + if not isinstance(body, bytes): + raise ProtocolError( + f'{self.alias}: message data of {len(data)} octets needs ' + f'{pad_len} octet(s) of padding, but the body is neither bytes ' + 'nor a schema carrying bytes, so there is nowhere to put it') + data.data = body + b'\x00' * pad_len + return data + + # NOTE: Rebinding the attribute rather than mutating the list in place is + # what marks the schema as updated, so that ``len()`` and ``pack()`` take + # the padding into account. + data.options = list(options) + pad_opts + return data + def _read_msg_unknown(self, schema: 'Schema_UnknownMessage', *, header: 'Schema_MH') -> 'Data_UnknownMessage': """Read unknown MH message type. @@ -7767,70 +7831,6 @@ def _make_pad_options(self, offset: 'int') -> 'tuple[list[Schema_PadOption], int return [self._make_opt_pad(Enum_Option.Pad1, length=0)], pad_len # type: ignore[arg-type] return [self._make_opt_pad(Enum_Option.PadN, length=pad_len - 2)], pad_len # type: ignore[arg-type] - def _pad_mh_message(self, data: 'Schema_Packet | bytes') -> 'Schema_Packet | bytes': - """Pad an MH message so that the Mobility Header aligns to 8 octets. - - Args: - data: Constructed message data. - - Returns: - The message data, with padding options appended if any were needed. - - Note: - The padding goes into the message's mobility options, which is where - :rfc:`6275#section-6.1.1` puts it. A message whose body is opaque has - nowhere to put it -- - :class:`~pcapkit.protocols.schema.internet.mh.UnknownMessage` and - :class:`~pcapkit.protocols.schema.internet.mh.ExperimentalMessage` - carry raw bytes and no options, as does a ``data`` argument given - directly as :obj:`bytes` -- so for those the padding is appended to the - message body itself and a - :class:`~pcapkit.utilities.warnings.ProtocolWarning` says so. - - Appending is necessary rather than optional: ``length`` is - ``(len(data) + 6) // 8 - 1``, which floors, so leaving an opaque body - short would emit 10, 12 or 14 octets while declaring 8, and a parser reads - 8 and misinterprets the remainder. Since the caller asked for a packet - to be built and the shortfall is recoverable, completing it beats - refusing -- the warning is there because the emitted body is then not - byte-for-byte what was handed in. - - """ - pad_opts, pad_len = self._make_pad_options(len(data) + 6) - if pad_len == 0: - return data - - options = getattr(data, 'options', None) - # NOTE: The ``isinstance`` test comes first so that the type checker can - # narrow ``data`` for the assignment below; at runtime ``getattr`` has - # already covered the :obj:`bytes` case by returning :obj:`None`. - if isinstance(data, bytes) or options is None: - warn(f'{self.alias}: message data of {len(data)} octets carries no ' - f'mobility options to hold padding, so {pad_len} octet(s) were ' - 'appended to the message body to align the header', - ProtocolWarning) - if isinstance(data, bytes): - return data + b'\x00' * pad_len - - # An opaque schema body -- UnknownMessage, ExperimentalMessage -- keeps - # its content in ``data`` rather than in options, so that is where the - # octets go. Rebound rather than mutated in place so that ``len()`` and - # ``pack()`` see the change, exactly as for the options branch below. - body = getattr(data, 'data', None) - if not isinstance(body, bytes): - raise ProtocolError( - f'{self.alias}: message data of {len(data)} octets needs ' - f'{pad_len} octet(s) of padding, but the body is neither bytes ' - 'nor a schema carrying bytes, so there is nowhere to put it') - data.data = body + b'\x00' * pad_len - return data - - # NOTE: Rebinding the attribute rather than mutating the list in place is - # what marks the schema as updated, so that ``len()`` and ``pack()`` take - # the padding into account. - data.options = list(options) + pad_opts - return data - def _make_mh_options(self, options: 'Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]') -> 'list[Schema_Option | bytes]': """Make options for MH.