diff --git a/pcapkit/protocols/application/ospf.py b/pcapkit/protocols/application/ospf.py index 15ff6b846c..8aae88b5b1 100644 --- a/pcapkit/protocols/application/ospf.py +++ b/pcapkit/protocols/application/ospf.py @@ -414,37 +414,6 @@ def _read_id_numbers(self, id: 'bytes') -> 'IPv4Address': #_addr = '.'.join(str(_) for _ in _byte) return ipaddress.ip_address(id) # type: ignore[return-value] - def _read_encrypt_auth(self, schema: 'Schema_CryptographicAuthentication') -> 'Data_CryptographicAuthentication': - """Read Authentication field when Cryptographic Authentication is employed, - i.e. :attr:`~pcapkit.protocols.data.application.ospf.OSPF.autype` is ``2``. - - Structure of Cryptographic Authentication [:rfc:`2328`]: - - .. code-block:: text - - 0 1 2 3 - 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - | 0 | Key ID | Auth Data Len | - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - | Cryptographic sequence number | - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - - Args: - schema: parsed authentication data - - Returns: - Parsed packet data. - - """ - auth = Data_CryptographicAuthentication( - reserved=schema.reserved, - key_id=schema.key_id, - len=schema.len, - seq=schema.seq, - ) - return auth - def _make_id_numbers(self, id: 'IPv4Address | str | bytes | bytearray') -> 'bytes': """Make router and area IDs. @@ -478,6 +447,37 @@ def _make_id_numbers(self, id: 'IPv4Address | str | bytes | bytearray') -> 'byte return cast('IPv4Address', parse_ip_address( id, f'{self.__class__.__name__}: invalid ID', version=4)).packed + def _read_encrypt_auth(self, schema: 'Schema_CryptographicAuthentication') -> 'Data_CryptographicAuthentication': + """Read Authentication field when Cryptographic Authentication is employed, + i.e. :attr:`~pcapkit.protocols.data.application.ospf.OSPF.autype` is ``2``. + + Structure of Cryptographic Authentication [:rfc:`2328`]: + + .. code-block:: text + + 0 1 2 3 + 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | 0 | Key ID | Auth Data Len | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Cryptographic sequence number | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Args: + schema: parsed authentication data + + Returns: + Parsed packet data. + + """ + auth = Data_CryptographicAuthentication( + reserved=schema.reserved, + key_id=schema.key_id, + len=schema.len, + seq=schema.seq, + ) + return auth + def _make_encrypt_auth(self, auth_data: 'bytes | Schema_CryptographicAuthentication | Data_CryptographicAuthentication | dict[str, Any]' # pylint: disable=line-too-long ) -> 'Schema_CryptographicAuthentication': diff --git a/pcapkit/protocols/internet/igmpv3.py b/pcapkit/protocols/internet/igmpv3.py index a81fd57e2c..5fdef5d440 100644 --- a/pcapkit/protocols/internet/igmpv3.py +++ b/pcapkit/protocols/internet/igmpv3.py @@ -407,6 +407,55 @@ def make(self, raise ProtocolError(f'{self.alias}: unsupported message type: {type_value!r}') + ########################################################################## + # Utilities. + ########################################################################## + + @classmethod + def _make_data(cls, data: 'Data_IGMPv3') -> 'dict[str, Any]': # type: ignore[override] + """Create key-value pairs from ``data`` for protocol construction. + + Args: + data: protocol data + + Returns: + Key-value pairs for protocol construction. + + Note: + Dispatches on ``data.type`` rather than on + :func:`isinstance(data, Data_Report) `, so that a + hand-built :obj:`dict` carrying the right ``type`` but not + actually one of the two concrete data classes -- as a test double + legitimately might -- still rebuilds correctly, the same way + :meth:`make` itself dispatches on the ``type`` it is given rather + than requiring a particular input class. + + """ + if data.type == Enum_Type.IGMPv3_Membership_Report: + report = cast('Data_Report', data) + return { + 'type': report.type, + 'reserved_1': report.reserved_1, + 'checksum': report.checksum, + 'reserved_2': report.reserved_2, + 'records': report.records, + 'payload': cls._make_payload(data), + } + + query = cast('Data_Query', data) + return { + 'type': query.type, + 'max_resp_code': query.max_resp_code, + 'checksum': query.checksum, + 'group': query.group, + 'resv': query.resv, + 's_flag': query.s_flag, + 'qrv': query.qrv, + 'qqic': query.qqic, + 'sources': query.sources, + 'payload': cls._make_payload(data), + } + @staticmethod def _make_group_record(record: 'Optional[Data_GroupRecord]' = None, *, record_type: 'Enum_RecordType | int' = Enum_RecordType.MODE_IS_INCLUDE, @@ -458,52 +507,3 @@ def _make_group_record(record: 'Optional[Data_GroupRecord]' = None, *, sources=source_list, aux_data=aux_data, ) - - ########################################################################## - # Utilities. - ########################################################################## - - @classmethod - def _make_data(cls, data: 'Data_IGMPv3') -> 'dict[str, Any]': # type: ignore[override] - """Create key-value pairs from ``data`` for protocol construction. - - Args: - data: protocol data - - Returns: - Key-value pairs for protocol construction. - - Note: - Dispatches on ``data.type`` rather than on - :func:`isinstance(data, Data_Report) `, so that a - hand-built :obj:`dict` carrying the right ``type`` but not - actually one of the two concrete data classes -- as a test double - legitimately might -- still rebuilds correctly, the same way - :meth:`make` itself dispatches on the ``type`` it is given rather - than requiring a particular input class. - - """ - if data.type == Enum_Type.IGMPv3_Membership_Report: - report = cast('Data_Report', data) - return { - 'type': report.type, - 'reserved_1': report.reserved_1, - 'checksum': report.checksum, - 'reserved_2': report.reserved_2, - 'records': report.records, - 'payload': cls._make_payload(data), - } - - query = cast('Data_Query', data) - return { - 'type': query.type, - 'max_resp_code': query.max_resp_code, - 'checksum': query.checksum, - 'group': query.group, - 'resv': query.resv, - 's_flag': query.s_flag, - 'qrv': query.qrv, - 'qqic': query.qqic, - 'sources': query.sources, - 'payload': cls._make_payload(data), - } diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index cb6fc8ff0f..99c76ddd87 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -820,6 +820,114 @@ def _read_ipv4_options(self, length: 'int') -> 'Option': raise ProtocolError(f'IPv4: invalid format') return options + def _make_ipv4_options(self, options: 'list[Schema_Option | tuple[Enum_OptionNumber, dict[str, Any]] | bytes] | Option', + padding: 'Optional[bytes]' = None) -> 'tuple[list[Schema_Option | bytes], int]': + """Make options for IPv4. + + A :obj:`list` is a fresh option list, so any ``NOP`` and ``EOOL`` in it + are dropped and the padding is generated. A container is what + :meth:`read` hands back, so its ``NOP`` and ``EOOL`` options are written + in place, as parsed. + + Either way, the options are concatenated and then padded once, after the + last one, to the 32-bit boundary: with ``NOP`` options and a closing + ``EOOL``, or with zero octets if an ``EOOL`` has already ended the list. + ``padding`` goes after that ``EOOL`` verbatim, before any alignment; it + is dropped if no ``EOOL`` ends the list, since the reader would take its + octets for options. + + Args: + options: IPv4 options + padding: octets after the End of Option List + + Returns: + Tuple of options and total length of options. + + """ + total_length = 0 + terminated = False + options_list = [] # type: list[Schema_Option | bytes] + if isinstance(options, list): + for schema in options: + if isinstance(schema, bytes): + code = Enum_OptionNumber.get(schema[0]) + if code in (Enum_OptionNumber.NOP, Enum_OptionNumber.EOOL): # ignore padding options by default + continue + + data = schema # type: Schema_Option | bytes + data_len = len(data) + elif isinstance(schema, Schema): + code = schema.type + if code in (Enum_OptionNumber.NOP, Enum_OptionNumber.EOOL): # ignore padding options by default + continue + + data = schema + data_len = len(schema.pack()) + else: + code, args = cast('tuple[Enum_OptionNumber, dict[str, Any]]', schema) + if code in (Enum_OptionNumber.NOP, Enum_OptionNumber.EOOL): # 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_unassigned)) + else: + meth = name[1] + + data = meth(code, **args) + data_len = len(data.pack()) + + options_list.append(data) + total_length += data_len + else: + 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_unassigned)) + else: + meth = name[1] + + data = meth(code, option) + data_len = len(data.pack()) + + options_list.append(data) + total_length += data_len + if code == Enum_OptionNumber.EOOL: + terminated = True + + # NOTE: Raw octets are safe here only after the EOOL: the reader + # stops there, and never looks at what follows as an option. + if terminated and padding: + options_list.append(bytes(padding)) + total_length += len(padding) + + # force alignment to 32-bit boundary + # + # NOTE: Once, after the last option. An EOOL after each unaligned + # option would end the list there, and every later option would be + # lost on parse. The terminator is an EOOL option *schema*, like the + # NOP padding: the ``options`` field takes only schemas and + # :obj:`bytes`, not the bare wire code. + if total_length % 4: + pad_len = 4 - (total_length % 4) + total_length += pad_len + + # zero octets after an EOOL already in the list, as EOOL schemas + if terminated: + pad_opt = cast('Schema_Option', self._make_opt_eool(Enum_OptionNumber.EOOL)) # type: ignore[arg-type] + else: + pad_opt = cast('Schema_Option', self._make_opt_nop(Enum_OptionNumber.NOP)) # type: ignore[arg-type] + end_opt = self._make_opt_eool(Enum_OptionNumber.EOOL) # type: ignore[arg-type] + + for _ in range(pad_len - 1): + options_list.append(pad_opt) + options_list.append(end_opt) + return options_list, total_length + def _read_opt_unassigned(self, schema: 'Schema_UnassignedOption', *, options: 'Option') -> 'Data_UnassignedOption': # pylint: disable=unused-argument """Read IPv4 unassigned options. @@ -1438,114 +1546,6 @@ def _read_opt_qs(self, schema: 'Schema_QSOption', *, options: 'Option') -> 'Data raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] unknown QS function: {func}') return opt - def _make_ipv4_options(self, options: 'list[Schema_Option | tuple[Enum_OptionNumber, dict[str, Any]] | bytes] | Option', - padding: 'Optional[bytes]' = None) -> 'tuple[list[Schema_Option | bytes], int]': - """Make options for IPv4. - - A :obj:`list` is a fresh option list, so any ``NOP`` and ``EOOL`` in it - are dropped and the padding is generated. A container is what - :meth:`read` hands back, so its ``NOP`` and ``EOOL`` options are written - in place, as parsed. - - Either way, the options are concatenated and then padded once, after the - last one, to the 32-bit boundary: with ``NOP`` options and a closing - ``EOOL``, or with zero octets if an ``EOOL`` has already ended the list. - ``padding`` goes after that ``EOOL`` verbatim, before any alignment; it - is dropped if no ``EOOL`` ends the list, since the reader would take its - octets for options. - - Args: - options: IPv4 options - padding: octets after the End of Option List - - Returns: - Tuple of options and total length of options. - - """ - total_length = 0 - terminated = False - options_list = [] # type: list[Schema_Option | bytes] - if isinstance(options, list): - for schema in options: - if isinstance(schema, bytes): - code = Enum_OptionNumber.get(schema[0]) - if code in (Enum_OptionNumber.NOP, Enum_OptionNumber.EOOL): # ignore padding options by default - continue - - data = schema # type: Schema_Option | bytes - data_len = len(data) - elif isinstance(schema, Schema): - code = schema.type - if code in (Enum_OptionNumber.NOP, Enum_OptionNumber.EOOL): # ignore padding options by default - continue - - data = schema - data_len = len(schema.pack()) - else: - code, args = cast('tuple[Enum_OptionNumber, dict[str, Any]]', schema) - if code in (Enum_OptionNumber.NOP, Enum_OptionNumber.EOOL): # 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_unassigned)) - else: - meth = name[1] - - data = meth(code, **args) - data_len = len(data.pack()) - - options_list.append(data) - total_length += data_len - else: - 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_unassigned)) - else: - meth = name[1] - - data = meth(code, option) - data_len = len(data.pack()) - - options_list.append(data) - total_length += data_len - if code == Enum_OptionNumber.EOOL: - terminated = True - - # NOTE: Raw octets are safe here only after the EOOL: the reader - # stops there, and never looks at what follows as an option. - if terminated and padding: - options_list.append(bytes(padding)) - total_length += len(padding) - - # force alignment to 32-bit boundary - # - # NOTE: Once, after the last option. An EOOL after each unaligned - # option would end the list there, and every later option would be - # lost on parse. The terminator is an EOOL option *schema*, like the - # NOP padding: the ``options`` field takes only schemas and - # :obj:`bytes`, not the bare wire code. - if total_length % 4: - pad_len = 4 - (total_length % 4) - total_length += pad_len - - # zero octets after an EOOL already in the list, as EOOL schemas - if terminated: - pad_opt = cast('Schema_Option', self._make_opt_eool(Enum_OptionNumber.EOOL)) # type: ignore[arg-type] - else: - pad_opt = cast('Schema_Option', self._make_opt_nop(Enum_OptionNumber.NOP)) # type: ignore[arg-type] - end_opt = self._make_opt_eool(Enum_OptionNumber.EOOL) # type: ignore[arg-type] - - for _ in range(pad_len - 1): - options_list.append(pad_opt) - options_list.append(end_opt) - return options_list, total_length - def _make_opt_unassigned(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_UnassignedOption]' = None, *, data: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_Option': diff --git a/pcapkit/protocols/internet/ipx.py b/pcapkit/protocols/internet/ipx.py index 42d0ed4e45..9b89985f88 100644 --- a/pcapkit/protocols/internet/ipx.py +++ b/pcapkit/protocols/internet/ipx.py @@ -324,22 +324,6 @@ def _next_exthdr_depth(self, proto: 'Optional[int]', extension: 'bool' = False) """ return 0 - @staticmethod - def _make_ipx_address(addr: 'Data_Address') -> 'bytes': - """Make IPX address field. - - Args: - addr: Parsed IPX address field. - - Returns: - The 12-octet IPX address: network number, node number and - socket number. - - """ - return (bytes.fromhex(addr.network.replace(':', '')) - + bytes.fromhex(addr.node.replace('-', '')) - + int(addr.socket).to_bytes(2, 'big', signed=False)) - def _read_ipx_address(self, addr: 'bytes') -> 'Data_Address': """Read IPX address field. @@ -371,3 +355,19 @@ def _read_ipx_address(self, addr: 'bytes') -> 'Data_Address': addr=_addr, ) return ipx_addr + + @staticmethod + def _make_ipx_address(addr: 'Data_Address') -> 'bytes': + """Make IPX address field. + + Args: + addr: Parsed IPX address field. + + Returns: + The 12-octet IPX address: network number, node number and + socket number. + + """ + return (bytes.fromhex(addr.network.replace(':', '')) + + bytes.fromhex(addr.node.replace('-', '')) + + int(addr.socket).to_bytes(2, 'big', signed=False)) diff --git a/pcapkit/protocols/misc/checksum/ip_pseudo.py b/pcapkit/protocols/misc/checksum/ip_pseudo.py index f9f093f266..cb26fe18cb 100644 --- a/pcapkit/protocols/misc/checksum/ip_pseudo.py +++ b/pcapkit/protocols/misc/checksum/ip_pseudo.py @@ -380,6 +380,11 @@ def _make_data(cls, data: 'Data_IP_Pseudo') -> 'dict[str, Any]': # type: ignore 'zero': data.zero, } + @staticmethod + def _is_ipv6(address: 'IPv4Address | IPv6Address') -> 'bool': + """Tell whether ``address`` selects the IPv6 layout.""" + return isinstance(address, ipaddress.IPv6Address) + def _source(self, ip: 'IPv4 | IPv6') -> 'IPv4Address | IPv6Address': """Read the pseudo-header source address from an IP layer. @@ -554,11 +559,6 @@ def _upper(self, ip: 'IPv4 | IPv6') -> 'tuple[Enum_TransType, int]': 'give protocol and upper_length') return code, before - @staticmethod - def _is_ipv6(address: 'IPv4Address | IPv6Address') -> 'bool': - """Tell whether ``address`` selects the IPv6 layout.""" - return isinstance(address, ipaddress.IPv6Address) - @staticmethod def _is_extension_header(code: 'int') -> 'bool': """Tell whether ``code`` is an IPv6 extension header number.""" diff --git a/pcapkit/protocols/schema/internet/igmpv3.py b/pcapkit/protocols/schema/internet/igmpv3.py index 7d97cf5e74..0dec9ffc8d 100644 --- a/pcapkit/protocols/schema/internet/igmpv3.py +++ b/pcapkit/protocols/schema/internet/igmpv3.py @@ -168,11 +168,6 @@ class Query(Schema): #: :attr:`QueryExtension.payload ` instead -- see there. payload: 'bytes' = PayloadField() - if TYPE_CHECKING: - def __init__(self, max_resp_code: 'int', checksum: 'bytes', # pylint: disable=unused-argument,super-init-not-called,multiple-statements - group: 'IPv4Address | str | bytes | int', ext: 'Optional[QueryExtension]', - payload: 'bytes | ProtocolBase | Schema') -> 'None': ... - def get_payload(self, name: 'Optional[str]' = None) -> 'bytes': """Get payload of :class:`Query`. @@ -190,6 +185,11 @@ def get_payload(self, name: 'Optional[str]' = None) -> 'bytes': return self.ext.get_payload() # pylint: disable=no-member return super().get_payload('payload') + if TYPE_CHECKING: + def __init__(self, max_resp_code: 'int', checksum: 'bytes', # pylint: disable=unused-argument,super-init-not-called,multiple-statements + group: 'IPv4Address | str | bytes | int', ext: 'Optional[QueryExtension]', + payload: 'bytes | ProtocolBase | Schema') -> 'None': ... + @schema_final class GroupRecord(Schema): @@ -253,11 +253,6 @@ class Report(Schema): #: :attr:`num_group_records` matching the octets actually present. payload: 'bytes' = PayloadField() - if TYPE_CHECKING: - def __init__(self, reserved_1: 'int', checksum: 'bytes', reserved_2: 'bytes', - num_group_records: 'int', records: 'list[GroupRecord]', - payload: 'bytes | ProtocolBase | Schema') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements - def post_process(self, packet: 'dict[str, Any]') -> 'Schema': # pylint: disable=unused-argument """Trim :attr:`records` to :attr:`num_group_records` and reclaim the rest as :attr:`payload`. @@ -287,6 +282,11 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': # pylint: disable self.__buffer__['payload'] = raw[consumed:] return self + if TYPE_CHECKING: + def __init__(self, reserved_1: 'int', checksum: 'bytes', reserved_2: 'bytes', + num_group_records: 'int', records: 'list[GroupRecord]', + payload: 'bytes | ProtocolBase | Schema') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements + @schema_final class IGMPv3(Schema): @@ -311,9 +311,6 @@ class IGMPv3(Schema): #: Message body, switched on :attr:`type`. body: 'Query | Report' = SwitchField(selector=igmpv3_body_selector) - if TYPE_CHECKING: - def __init__(self, type: 'Enum_Type | int', body: 'Query | Report | bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin - def get_payload(self, name: 'Optional[str]' = None) -> 'bytes': # pylint: disable=unused-argument """Get payload of :class:`IGMPv3`. @@ -329,3 +326,6 @@ def get_payload(self, name: 'Optional[str]' = None) -> 'bytes': # pylint: disab if isinstance(self.body, Schema): return self.body.get_payload() return b'' + + if TYPE_CHECKING: + def __init__(self, type: 'Enum_Type | int', body: 'Query | Report | bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index 94f725c253..1b51c88f80 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -976,43 +976,6 @@ def _get_payload(self) -> 'bytes': """ return self._payload - def _decode_next_layer(self, dict_: 'Data_SCTP', proto: 'Optional[int]' = None, # type: ignore[override] - length: 'Optional[int]' = None, *, - packet: 'Optional[dict[str, Any]]' = None) -> 'Data_SCTP': - r"""Decode next layer protocol. - - Arguments: - dict\_: info buffer - proto: payload protocol identifier of the DATA chunk carrying the - payload, if any - length: valid (*non-padding*) length - packet: packet info (passed from :meth:`self.unpack `) - - Returns: - Current protocol with next layer extracted. - - Important: - This deliberately bypasses :meth:`Transport._decode_next_layer - `, - which keys the lookup on port numbers, since SCTP keys it on the - DATA chunk's payload protocol identifier instead. - - The PPID is passed through **unchanged**, registered or not, so that - an unregistered payload is still labelled with the identifier it - arrived with, as :meth:`Internet._import_next_layer - ` - does for an unregistered transport type. Resolving it to - :class:`~pcapkit.protocols.misc.raw.Raw` is - :meth:`Protocol._import_next_layer - `'s job, - through :meth:`Protocol._lookup_next_layer - `, which - leaves :attr:`self.__proto__ ` untouched. - - """ - return ProtocolBase._decode_next_layer( # pylint: disable=protected-access - self, dict_, proto, length, packet=packet) # type: ignore[arg-type] - def _read_sctp_chunks(self) -> 'Chunks': """Read SCTP chunk list. @@ -1082,6 +1045,43 @@ def _make_sctp_chunks(self, chunks: 'list[Schema_Chunk | tuple[Enum_Chunk, dict[ chunks_list.append(self._make_sctp_chunk(code, chunk)) return _last_item_only(chunks_list) + def _decode_next_layer(self, dict_: 'Data_SCTP', proto: 'Optional[int]' = None, # type: ignore[override] + length: 'Optional[int]' = None, *, + packet: 'Optional[dict[str, Any]]' = None) -> 'Data_SCTP': + r"""Decode next layer protocol. + + Arguments: + dict\_: info buffer + proto: payload protocol identifier of the DATA chunk carrying the + payload, if any + length: valid (*non-padding*) length + packet: packet info (passed from :meth:`self.unpack `) + + Returns: + Current protocol with next layer extracted. + + Important: + This deliberately bypasses :meth:`Transport._decode_next_layer + `, + which keys the lookup on port numbers, since SCTP keys it on the + DATA chunk's payload protocol identifier instead. + + The PPID is passed through **unchanged**, registered or not, so that + an unregistered payload is still labelled with the identifier it + arrived with, as :meth:`Internet._import_next_layer + ` + does for an unregistered transport type. Resolving it to + :class:`~pcapkit.protocols.misc.raw.Raw` is + :meth:`Protocol._import_next_layer + `'s job, + through :meth:`Protocol._lookup_next_layer + `, which + leaves :attr:`self.__proto__ ` untouched. + + """ + return ProtocolBase._decode_next_layer( # pylint: disable=protected-access + self, dict_, proto, length, packet=packet) # type: ignore[arg-type] + def _make_sctp_chunk(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Chunk]' = None, **kwargs: 'Any') -> 'Schema_Chunk': """Dispatch to the chunk constructor registered for ``code``. diff --git a/pcapkit/protocols/transport/tcp.py b/pcapkit/protocols/transport/tcp.py index 25daebde9e..fc429949d8 100644 --- a/pcapkit/protocols/transport/tcp.py +++ b/pcapkit/protocols/transport/tcp.py @@ -903,6 +903,102 @@ def _read_tcp_options(self, size: 'int') -> 'Option': raise ProtocolError('TCP: invalid format') return options + def _make_tcp_options(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option', + padding: 'bytes' = b'') -> 'tuple[list[Schema_Option | bytes], int]': + """Make options for TCP. + + Args: + options: TCP options + padding: octets after the End of Option List option + + Returns: + Tuple of options and total length of options. + + Raises: + ProtocolError: If an option maker rejects its option, e.g. a user + timeout too large for 15 bits of minutes, or an MPTCP option + whose fields, or the segment's flags, it cannot encode. + ProtocolNotImplemented: If an option argument given by name, an + MPTCP ``subtype`` or an alternate checksum ``algorithm``, is + not in its namespace and has no default. + FieldValueError: If an ADD_ADDR ``addr`` is a :obj:`bool` or not a + valid IP address. + + Note: + The options are emitted in the given order, ``NOP`` and ``EOOL`` + included, so that a parsed option list re-emits the same octets. + ``padding`` follows them as given if they end with ``EOOL``, and is + dropped otherwise. Alignment is added only once, after that, when + the list does not end on a 32-bit boundary: zero octets if the list + already ends with ``EOOL``, otherwise ``NOP`` options followed by an + ``EOOL``. + + """ + total_length = 0 + options_list = [] # type: list[Schema_Option | bytes] + code = None # type: Optional[Enum_Option] + if isinstance(options, list): + for schema in options: + if isinstance(schema, bytes): + code = Enum_Option.get(schema[0]) + data = schema # type: Schema_Option | bytes + data_len = len(data) + elif isinstance(schema, Schema): + code = schema.kind + data = schema + data_len = len(schema.pack()) + else: + code, args = cast('tuple[Enum_Option, dict[str, Any]]', schema) + name = self._lookup_registry(self.__option__, code) + if isinstance(name, str): + meth_name = f'_make_mode_{name}' + meth = cast('OptionConstructor', + getattr(self, meth_name, self._make_mode_donone)) + else: + meth = name[1] + + data = meth(code, **args) + data_len = len(data.pack()) + + options_list.append(data) + total_length += data_len + else: + for code, option in options.items(multi=True): + name = self._lookup_registry(self.__option__, code) + if isinstance(name, str): + meth_name = f'_make_mode_{name}' + meth = cast('OptionConstructor', + getattr(self, meth_name, self._make_mode_donone)) + else: + meth = name[1] + + data = meth(code, option) + data_len = len(data.pack()) + + options_list.append(data) + total_length += data_len + + # NOTE: ``padding`` belongs after an End of Option List option; with + # that gone -- e.g. options edited after parsing -- it is dropped, and + # the Data Offset is computed from the options alone. + if padding and code == Enum_Option.End_of_Option_List: + options_list.append(padding) + total_length += len(padding) + + # force alignment to 32-bit boundary + if total_length % 4: + pad_len = 4 - (total_length % 4) + total_length += pad_len + + if code == Enum_Option.End_of_Option_List: + options_list.append(bytes(pad_len)) + else: + pad_opt = self._make_mode_nop(Enum_Option.No_Operation) # type: ignore[arg-type] + for _ in range(pad_len - 1): + options_list.append(pad_opt) + options_list.append(self._make_mode_eool(Enum_Option.End_of_Option_List)) # type: ignore[arg-type] + return options_list, total_length + def _read_mode_donone(self, schema: 'Schema_UnassignedOption', *, options: 'Option') -> 'Data_UnassignedOption': # pylint: disable=unused-argument """Read TCP unassigned option. @@ -2216,102 +2312,6 @@ def _read_mode_fastopen(self, schema: 'Schema_FastOpenCookie', *, options: 'Opti ) return data - def _make_tcp_options(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option', - padding: 'bytes' = b'') -> 'tuple[list[Schema_Option | bytes], int]': - """Make options for TCP. - - Args: - options: TCP options - padding: octets after the End of Option List option - - Returns: - Tuple of options and total length of options. - - Raises: - ProtocolError: If an option maker rejects its option, e.g. a user - timeout too large for 15 bits of minutes, or an MPTCP option - whose fields, or the segment's flags, it cannot encode. - ProtocolNotImplemented: If an option argument given by name, an - MPTCP ``subtype`` or an alternate checksum ``algorithm``, is - not in its namespace and has no default. - FieldValueError: If an ADD_ADDR ``addr`` is a :obj:`bool` or not a - valid IP address. - - Note: - The options are emitted in the given order, ``NOP`` and ``EOOL`` - included, so that a parsed option list re-emits the same octets. - ``padding`` follows them as given if they end with ``EOOL``, and is - dropped otherwise. Alignment is added only once, after that, when - the list does not end on a 32-bit boundary: zero octets if the list - already ends with ``EOOL``, otherwise ``NOP`` options followed by an - ``EOOL``. - - """ - total_length = 0 - options_list = [] # type: list[Schema_Option | bytes] - code = None # type: Optional[Enum_Option] - if isinstance(options, list): - for schema in options: - if isinstance(schema, bytes): - code = Enum_Option.get(schema[0]) - data = schema # type: Schema_Option | bytes - data_len = len(data) - elif isinstance(schema, Schema): - code = schema.kind - data = schema - data_len = len(schema.pack()) - else: - code, args = cast('tuple[Enum_Option, dict[str, Any]]', schema) - name = self._lookup_registry(self.__option__, code) - if isinstance(name, str): - meth_name = f'_make_mode_{name}' - meth = cast('OptionConstructor', - getattr(self, meth_name, self._make_mode_donone)) - else: - meth = name[1] - - data = meth(code, **args) - data_len = len(data.pack()) - - options_list.append(data) - total_length += data_len - else: - for code, option in options.items(multi=True): - name = self._lookup_registry(self.__option__, code) - if isinstance(name, str): - meth_name = f'_make_mode_{name}' - meth = cast('OptionConstructor', - getattr(self, meth_name, self._make_mode_donone)) - else: - meth = name[1] - - data = meth(code, option) - data_len = len(data.pack()) - - options_list.append(data) - total_length += data_len - - # NOTE: ``padding`` belongs after an End of Option List option; with - # that gone -- e.g. options edited after parsing -- it is dropped, and - # the Data Offset is computed from the options alone. - if padding and code == Enum_Option.End_of_Option_List: - options_list.append(padding) - total_length += len(padding) - - # force alignment to 32-bit boundary - if total_length % 4: - pad_len = 4 - (total_length % 4) - total_length += pad_len - - if code == Enum_Option.End_of_Option_List: - options_list.append(bytes(pad_len)) - else: - pad_opt = self._make_mode_nop(Enum_Option.No_Operation) # type: ignore[arg-type] - for _ in range(pad_len - 1): - options_list.append(pad_opt) - options_list.append(self._make_mode_eool(Enum_Option.End_of_Option_List)) # type: ignore[arg-type] - return options_list, total_length - def _make_mode_donone(self, code: 'Enum_Option', opt: 'Optional[Data_UnassignedOption]' = None, *, data: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_UnassignedOption': diff --git a/tests/project/test_protocol_member_order_2490_unit.py b/tests/project/test_protocol_member_order_2490_unit.py index d5179da68e..f304f1104c 100644 --- a/tests/project/test_protocol_member_order_2490_unit.py +++ b/tests/project/test_protocol_member_order_2490_unit.py @@ -66,20 +66,8 @@ class no longer breaks, so a reorder deletes its entry in the same change. #: Classes that break the order, as ``qualified name: (rules, remover)``: the #: :data:`RULES` kinds each breaks, and the issue or PR whose reorder removes the -#: entry. ``#2490`` marks a file deferred from its reorder while another stream -#: edits it; IPv4 and TCP wait for the rule c reorders (#2490). -KNOWN_EXCEPTIONS = { - 'pcapkit.protocols.application.ospf.OSPF': (frozenset({'helpers'}), '#2490'), - 'pcapkit.protocols.internet.igmpv3.IGMPv3': (frozenset({'data'}), '#2490'), - 'pcapkit.protocols.internet.ipv4.IPv4': (frozenset({'helpers'}), '#2490'), - 'pcapkit.protocols.internet.ipx.IPX': (frozenset({'helpers', 'pair'}), '#2490'), - 'pcapkit.protocols.misc.checksum.ip_pseudo.IP_Pseudo': (frozenset({'helpers'}), '#2490'), - 'pcapkit.protocols.schema.internet.igmpv3.IGMPv3': (frozenset({'group'}), '#2490'), - 'pcapkit.protocols.schema.internet.igmpv3.Query': (frozenset({'group'}), '#2490'), - 'pcapkit.protocols.schema.internet.igmpv3.Report': (frozenset({'group'}), '#2490'), - 'pcapkit.protocols.transport.sctp.SCTP': (frozenset({'helpers'}), '#2490'), - 'pcapkit.protocols.transport.tcp.TCP': (frozenset({'helpers'}), '#2490'), -} # type: dict[str, tuple[frozenset[str], str]] +#: entry. +KNOWN_EXCEPTIONS = {} # type: dict[str, tuple[frozenset[str], str]] _DEFS = (ast.FunctionDef, ast.AsyncFunctionDef)