From 6a949ca7e7aacddcbcf1365e7e01232369726ec2 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sun, 1 Nov 2026 10:56:16 -0500 Subject: [PATCH] refactor(protocols): reorder misc, schema and base class members to the member-order convention (#2490) - protocol: move _packing_target and _next_layer_entry out of Methods, and _make_data to the head of Utilities. - pcap frame/header: move Frame.__index__ into Data models and its private helpers into Utilities after _make_data; move Header.from_data into Methods. - pcapng: move _make_data to the head of Utilities, then the helpers read() calls directly, in first-use order (_read_block_short with its maker). - schema: put Methods ahead of Data models in Schema, EnumSchema, Option, FlagsField, OptionEnumField and SystemdJournalExportBlock; put the IPv6 schema's TYPE_CHECKING __init__ stub after post_process. Pure move: member set, member ASTs and comments are unchanged. --- pcapkit/protocols/misc/pcap/frame.py | 112 +-- pcapkit/protocols/misc/pcap/header.py | 32 +- pcapkit/protocols/misc/pcapng.py | 344 +++---- pcapkit/protocols/protocol.py | 176 ++-- pcapkit/protocols/schema/internet/ipv6.py | 14 +- pcapkit/protocols/schema/misc/pcapng.py | 218 ++--- pcapkit/protocols/schema/schema.py | 1082 ++++++++++----------- 7 files changed, 989 insertions(+), 989 deletions(-) diff --git a/pcapkit/protocols/misc/pcap/frame.py b/pcapkit/protocols/misc/pcap/frame.py index 64093feb54..2bf67003d7 100644 --- a/pcapkit/protocols/misc/pcap/frame.py +++ b/pcapkit/protocols/misc/pcap/frame.py @@ -521,36 +521,6 @@ def __length_hint__(self) -> 'Literal[16]': """Return an estimated length for the object.""" return RECORD_HEADER_LEN # type: ignore[return-value] - def _check_record_floor(self, length: 'Optional[int]') -> 'None': - """Refuse a tail too short to hold a record header. - - Args: - length: Length of packet data, as the caller declared it, or - :obj:`None` to measure what is left in - :attr:`self._file `. - - Raises: - StreamEOFError: If fewer than sixteen octets are left to read. - - A PCAP record header is sixteen octets, which is what - :meth:`__length_hint__` reports. A shorter tail is not a record, so - parsing one would only invent fields from zero padding. Reporting the - end of the stream instead lets :meth:`Extractor.record_frames - ` stop with the - frames read so far, as the PCAP-NG reader does for a truncated block - (:issue:`678`). See :issue:`1265`. - - """ - if length is None: - current = self._file.tell() - length = self._file.seek(0, io.SEEK_END) - current - self._file.seek(current, io.SEEK_SET) - - if length < RECORD_HEADER_LEN: - raise StreamEOFError(f'PCAP: record truncated: {length} octet(s) left, ' - f'fewer than the {RECORD_HEADER_LEN} a record header needs', - quiet=True) - # NOTE: This is a hack to make the ``__index__`` method work both as a # class method and an instance method. def __index__(self: 'Optional[Frame]' = None) -> 'int': # type: ignore[override] @@ -601,38 +571,35 @@ def _make_data(cls, data: 'Data_Frame') -> 'dict[str, Any]': # type: ignore[ove 'packet': cls._make_payload(data), } - def _make_timestamp(self, timestamp: 'Optional[float | Decimal | dt_type | int]' = None, ts_sec: 'Optional[int]' = None, - ts_usec: 'Optional[int]' = None, nanosecond: 'bool' = False) -> 'tuple[int, int]': - """Make timestamp. + def _check_record_floor(self, length: 'Optional[int]') -> 'None': + """Refuse a tail too short to hold a record header. Args: - timestamp: UNIX-Epoch timestamp - ts_sec: timestamp seconds - ts_usec: timestamp microseconds - nanosecond: nanosecond-resolution file flag - - Returns: - Second and microsecond/nanosecond value of timestamp. + length: Length of packet data, as the caller declared it, or + :obj:`None` to measure what is left in + :attr:`self._file `. - """ - with localcontext(prec=64): - if timestamp is None: - if nanosecond: - timestamp = decimal.Decimal(time.time_ns()) / 1_000_000_000 - else: - timestamp = decimal.Decimal(time.time()) - else: - if isinstance(timestamp, datetime.datetime): - timestamp = totimestamp(timestamp) - timestamp = decimal.Decimal(timestamp) + Raises: + StreamEOFError: If fewer than sixteen octets are left to read. - if ts_sec is None: - ts_sec = int(timestamp) + A PCAP record header is sixteen octets, which is what + :meth:`__length_hint__` reports. A shorter tail is not a record, so + parsing one would only invent fields from zero padding. Reporting the + end of the stream instead lets :meth:`Extractor.record_frames + ` stop with the + frames read so far, as the PCAP-NG reader does for a truncated block + (:issue:`678`). See :issue:`1265`. - if ts_usec is None: - ts_usec = int((timestamp - ts_sec) * RESOLUTION[bool(nanosecond)]) + """ + if length is None: + current = self._file.tell() + length = self._file.seek(0, io.SEEK_END) - current + self._file.seek(current, io.SEEK_SET) - return ts_sec, ts_usec + if length < RECORD_HEADER_LEN: + raise StreamEOFError(f'PCAP: record truncated: {length} octet(s) left, ' + f'fewer than the {RECORD_HEADER_LEN} a record header needs', + quiet=True) def _decode_next_layer(self, dict_: 'Data_Frame', proto: 'Optional[int]' = None, length: 'Optional[int]' = None, *, packet: 'Optional[dict[str, Any]]' = None) -> 'Data_Frame': # pylint: disable=arguments-differ @@ -680,6 +647,39 @@ def _decode_next_layer(self, dict_: 'Data_Frame', proto: 'Optional[int]' = None, self._protos = chain # pylint: disable=attribute-defined-outside-init return dict_ + def _make_timestamp(self, timestamp: 'Optional[float | Decimal | dt_type | int]' = None, ts_sec: 'Optional[int]' = None, + ts_usec: 'Optional[int]' = None, nanosecond: 'bool' = False) -> 'tuple[int, int]': + """Make timestamp. + + Args: + timestamp: UNIX-Epoch timestamp + ts_sec: timestamp seconds + ts_usec: timestamp microseconds + nanosecond: nanosecond-resolution file flag + + Returns: + Second and microsecond/nanosecond value of timestamp. + + """ + with localcontext(prec=64): + if timestamp is None: + if nanosecond: + timestamp = decimal.Decimal(time.time_ns()) / 1_000_000_000 + else: + timestamp = decimal.Decimal(time.time()) + else: + if isinstance(timestamp, datetime.datetime): + timestamp = totimestamp(timestamp) + timestamp = decimal.Decimal(timestamp) + + if ts_sec is None: + ts_sec = int(timestamp) + + if ts_usec is None: + ts_usec = int((timestamp - ts_sec) * RESOLUTION[bool(nanosecond)]) + + return ts_sec, ts_usec + def _dispatch_linktype(self, proto: 'Optional[int]', length: 'Optional[int]' = None) -> 'Optional[int]': """Link type a frame's data is dispatched under. diff --git a/pcapkit/protocols/misc/pcap/header.py b/pcapkit/protocols/misc/pcap/header.py index f8cba67577..51fde30bb0 100644 --- a/pcapkit/protocols/misc/pcap/header.py +++ b/pcapkit/protocols/misc/pcap/header.py @@ -230,6 +230,22 @@ def make(self, reversed=network_reversed, pack=False), ) + @classmethod + def from_data(cls, data: 'Data_Header | dict[str, Any]', **kwargs: 'Any') -> 'Self': + """Create protocol instance from data. + + Args: + data: Protocol data. + **kwargs: Construction keywords, as for :meth:`Protocol.from_data + `. + + Returns: + Protocol instance, cut back to what was captured when the header + it was parsed from was cut short (:issue:`1470`). + + """ + return replay_short_read(super().from_data(data, **kwargs), data) + ########################################################################## # Data models. ########################################################################## @@ -281,22 +297,6 @@ def __length_hint__(self) -> 'Literal[24]': """Return an estimated length for the object.""" return GLOBAL_HEADER_LEN # type: ignore[return-value] - @classmethod - def from_data(cls, data: 'Data_Header | dict[str, Any]', **kwargs: 'Any') -> 'Self': - """Create protocol instance from data. - - Args: - data: Protocol data. - **kwargs: Construction keywords, as for :meth:`Protocol.from_data - `. - - Returns: - Protocol instance, cut back to what was captured when the header - it was parsed from was cut short (:issue:`1470`). - - """ - return replay_short_read(super().from_data(data, **kwargs), data) - @classmethod def __index__(cls) -> 'NoReturn': """Numeral registry index of the protocol. diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index 8f24d206bb..c94c712442 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -1654,6 +1654,22 @@ def __index__(self: 'Optional[PCAPNG]' = None) -> 'int': # type: ignore[overrid # Utilities. ########################################################################## + @classmethod + def _make_data(cls, data: 'Data_PCAPNG') -> '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. + + """ + return { + 'type': data.type, + 'block': data, + } + def _check_block_floor(self, length: 'Optional[int]') -> 'None': """Reject a tail too short to hold any block at all. @@ -1715,6 +1731,162 @@ def _check_block_floor(self, length: 'Optional[int]') -> 'None': raise StreamEOFError(f'PCAP-NG: block truncated: {length} octet(s) left, ' f'fewer than the {GENERIC_BLOCK_MINIMUM} a block needs', quiet=True) + def _read_block_short(self, schema: 'Schema_EnhancedPacketBlock | Schema_PacketBlock | Schema_UnknownBlock', *, + header: 'Schema_PCAPNG') -> 'Data_EnhancedPacketBlock | Data_PacketBlock': + """Read a PCAP-NG EPB or packet block (obsolete) shorter than its fixed fields. + + The fixed fields of either block span 32 octets, so a smaller Block + Total Length cannot hold them. + + Args: + schema: Parsed block schema, or on the construction path the + :class:`~pcapkit.protocols.schema.misc.pcapng.UnknownBlock` that + :meth:`_make_block_short` built. + header: Parsed PCAP-NG header schema. + + Returns: + Parsed packet data, with the fixed fields the block holds and zero + for those it does not, no packet data and no options. The block body + is kept as ``block_raw``, which the block's ``make`` writes back + verbatim, so the block rebuilds at the length it declared + (:issue:`1414`). + + """ + if isinstance(schema, Schema_UnknownBlock): + body = schema.body + else: + # the octets unpacked for the fixed fields, which start with the + # Block Total Length and run past the end of the block + octets = bytes(schema) + body = octets[4:schema.length - 8] + warn(f'PCAP-NG: [Block {header.type}] block length {schema.length} is below the ' + f'32-octet minimum; kept as captured', ProtocolWarning, stacklevel=stacklevel()) + if octets[schema.length - 8:schema.length - 4] != octets[:4]: + warn(f'PCAP-NG: [Block {header.type}] block length mismatch: {schema.length} != ' + f'{octets[schema.length - 8:schema.length - 4].hex()}', ProtocolWarning, + stacklevel=stacklevel()) + + endian = '<' if self._byte == 'little' else '>' + fields = (body + bytes(20))[:20] + drop_count = 0 + if header.type == Enum_BlockType.Packet_Block: + interface_id, drop_count, timestamp_high, timestamp_low, captured_len, original_len = \ + struct.unpack(f'{endian}HHIIII', fields) + else: + interface_id, timestamp_high, timestamp_low, captured_len, original_len = \ + struct.unpack(f'{endian}IIIII', fields) + timestamp, timestamp_epoch = self._read_timestamp(timestamp_high, timestamp_low, + interface_id=interface_id) + + kwargs = { + 'type': header.type, + 'length': schema.length, + 'section_number': self._sect, + 'number': self._fnum, + 'interface_id': interface_id, + 'timestamp': timestamp, + 'timestamp_epoch': timestamp_epoch, + 'captured_len': captured_len, + 'original_len': original_len, + 'options': self._read_pcapng_options([]), + } # type: dict[str, Any] + data = (Data_PacketBlock(drop_count=drop_count, **kwargs) + if header.type == Enum_BlockType.Packet_Block + else Data_EnhancedPacketBlock(**kwargs)) # type: Data_EnhancedPacketBlock | Data_PacketBlock + data.__update__([('block_raw', body)]) + return self._decode_next_layer(data, self._get_linktype(interface_id), 0) # type: ignore[return-value] + + def _make_block_short( + self, block: 'Data_EnhancedPacketBlock | Data_PacketBlock | Data_SectionHeaderBlock | Data_InterfaceDescriptionBlock', # pylint: disable=line-too-long + ) -> 'Schema_UnknownBlock': + """Make a PCAP-NG EPB, packet block (obsolete), SHB or IDB kept as captured. + + Args: + block: Block data model, whose ``block_raw`` holds the body of a + block shorter than its fixed fields (see :meth:`_read_block_short` + and :meth:`_read_block_short_header`). + + Returns: + Constructed block schema, the body written back verbatim. + + """ + return self._make_block_unknown(data=block.block_raw) # type: ignore[union-attr] + + def _read_block_short_header( + self, schema: 'Schema_SectionHeaderBlock | Schema_InterfaceDescriptionBlock | Schema_UnknownBlock', *, + header: 'Schema_PCAPNG', + ) -> 'Data_SectionHeaderBlock | Data_InterfaceDescriptionBlock': + """Read a PCAP-NG SHB or IDB shorter than its fixed fields. + + The fixed fields of a Section Header Block span 28 octets and those of + an Interface Description Block 20 (see :attr:`SHORT_HEADER_TYPES`), so a + smaller Block Total Length cannot hold them. + + Args: + schema: Parsed block schema, or on the construction path the + :class:`~pcapkit.protocols.schema.misc.pcapng.UnknownBlock` that + :meth:`_make_block_short` built. + header: Parsed PCAP-NG header schema. + + Returns: + Parsed packet data, with the fixed fields the block holds and zero + for those it does not, and no options. The block body is kept as + ``block_raw``, which the block's ``make`` writes back verbatim, so + the block rebuilds at the length it declared (:issue:`1422`). + + A section header keeps the byte order its Byte-Order Magic declares. + One of 12 octets has no room for the magic, and takes the byte order + in which its Block Total Length reads as 12 (see + :class:`~pcapkit.protocols.schema.misc.pcapng.SectionHeaderBlock`). + + """ + if isinstance(schema, Schema_UnknownBlock): + body = schema.body + else: + # the octets unpacked for the fixed fields, which start with the + # Block Total Length and run past the end of the block + octets = bytes(schema) + body = octets[4:schema.length - 8] + warn(f'PCAP-NG: [Block {header.type}] block length {schema.length} is below the ' + f'{self.SHORT_HEADER_TYPES[header.type]}-octet minimum; kept as captured', + ProtocolWarning, stacklevel=stacklevel()) + if octets[schema.length - 8:schema.length - 4] != octets[:4]: + warn(f'PCAP-NG: [Block {header.type}] block length mismatch: {schema.length} != ' + f'{octets[schema.length - 8:schema.length - 4].hex()}', ProtocolWarning, + stacklevel=stacklevel()) + + data: 'Data_SectionHeaderBlock | Data_InterfaceDescriptionBlock' + if header.type == Enum_BlockType.Section_Header_Block: + # NOTE: On the construction path :meth:`_make_block_shb` has already + # set the byte order from the data model. + if isinstance(schema, Schema_SectionHeaderBlock): + self._byte = schema.byteorder + endian = '<' if self._byte == 'little' else '>' + major, minor, section_length = struct.unpack(f'{endian}HHq', (body + bytes(16))[4:16]) + data = Data_SectionHeaderBlock( + type=header.type, + length=schema.length, + byteorder=self._byte, + version=VersionInfo( + major=major, + minor=minor, + ), + section_length=section_length, + options=self._read_pcapng_options([]), + ) + else: + endian = '<' if self._byte == 'little' else '>' + linktype, _, snaplen = struct.unpack(f'{endian}HHI', (body + bytes(8))[:8]) + data = Data_InterfaceDescriptionBlock( + type=header.type, + length=schema.length, + linktype=Enum_LinkType.get(linktype), + snaplen=snaplen, + options=self._read_pcapng_options([]), + ) + data.__update__([('block_raw', body)]) + return data + def _captured(self, offset: 'int') -> 'int': """Count the octets left in the stream from ``offset`` on. @@ -1996,22 +2168,6 @@ def _read_timestamp(self, timestamp_high: 'int', timestamp_low: 'int', *, return (ts_datetime, timestamp_epoch) - @classmethod - def _make_data(cls, data: 'Data_PCAPNG') -> '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. - - """ - return { - 'type': data.type, - 'block': data, - } - @staticmethod def _block_area(block: 'Data_PCAPNG', name: 'str') -> 'Any': """Option or record area of a parsed block, for its ``make``. @@ -2918,146 +3074,6 @@ def _read_block_packet(self, schema: 'Schema_PacketBlock', *, return self._decode_next_layer(data, self._get_linktype(schema.interface_id), schema.captured_length) # type: ignore[return-value] - def _read_block_short(self, schema: 'Schema_EnhancedPacketBlock | Schema_PacketBlock | Schema_UnknownBlock', *, - header: 'Schema_PCAPNG') -> 'Data_EnhancedPacketBlock | Data_PacketBlock': - """Read a PCAP-NG EPB or packet block (obsolete) shorter than its fixed fields. - - The fixed fields of either block span 32 octets, so a smaller Block - Total Length cannot hold them. - - Args: - schema: Parsed block schema, or on the construction path the - :class:`~pcapkit.protocols.schema.misc.pcapng.UnknownBlock` that - :meth:`_make_block_short` built. - header: Parsed PCAP-NG header schema. - - Returns: - Parsed packet data, with the fixed fields the block holds and zero - for those it does not, no packet data and no options. The block body - is kept as ``block_raw``, which the block's ``make`` writes back - verbatim, so the block rebuilds at the length it declared - (:issue:`1414`). - - """ - if isinstance(schema, Schema_UnknownBlock): - body = schema.body - else: - # the octets unpacked for the fixed fields, which start with the - # Block Total Length and run past the end of the block - octets = bytes(schema) - body = octets[4:schema.length - 8] - warn(f'PCAP-NG: [Block {header.type}] block length {schema.length} is below the ' - f'32-octet minimum; kept as captured', ProtocolWarning, stacklevel=stacklevel()) - if octets[schema.length - 8:schema.length - 4] != octets[:4]: - warn(f'PCAP-NG: [Block {header.type}] block length mismatch: {schema.length} != ' - f'{octets[schema.length - 8:schema.length - 4].hex()}', ProtocolWarning, - stacklevel=stacklevel()) - - endian = '<' if self._byte == 'little' else '>' - fields = (body + bytes(20))[:20] - drop_count = 0 - if header.type == Enum_BlockType.Packet_Block: - interface_id, drop_count, timestamp_high, timestamp_low, captured_len, original_len = \ - struct.unpack(f'{endian}HHIIII', fields) - else: - interface_id, timestamp_high, timestamp_low, captured_len, original_len = \ - struct.unpack(f'{endian}IIIII', fields) - timestamp, timestamp_epoch = self._read_timestamp(timestamp_high, timestamp_low, - interface_id=interface_id) - - kwargs = { - 'type': header.type, - 'length': schema.length, - 'section_number': self._sect, - 'number': self._fnum, - 'interface_id': interface_id, - 'timestamp': timestamp, - 'timestamp_epoch': timestamp_epoch, - 'captured_len': captured_len, - 'original_len': original_len, - 'options': self._read_pcapng_options([]), - } # type: dict[str, Any] - data = (Data_PacketBlock(drop_count=drop_count, **kwargs) - if header.type == Enum_BlockType.Packet_Block - else Data_EnhancedPacketBlock(**kwargs)) # type: Data_EnhancedPacketBlock | Data_PacketBlock - data.__update__([('block_raw', body)]) - return self._decode_next_layer(data, self._get_linktype(interface_id), 0) # type: ignore[return-value] - - def _read_block_short_header( - self, schema: 'Schema_SectionHeaderBlock | Schema_InterfaceDescriptionBlock | Schema_UnknownBlock', *, - header: 'Schema_PCAPNG', - ) -> 'Data_SectionHeaderBlock | Data_InterfaceDescriptionBlock': - """Read a PCAP-NG SHB or IDB shorter than its fixed fields. - - The fixed fields of a Section Header Block span 28 octets and those of - an Interface Description Block 20 (see :attr:`SHORT_HEADER_TYPES`), so a - smaller Block Total Length cannot hold them. - - Args: - schema: Parsed block schema, or on the construction path the - :class:`~pcapkit.protocols.schema.misc.pcapng.UnknownBlock` that - :meth:`_make_block_short` built. - header: Parsed PCAP-NG header schema. - - Returns: - Parsed packet data, with the fixed fields the block holds and zero - for those it does not, and no options. The block body is kept as - ``block_raw``, which the block's ``make`` writes back verbatim, so - the block rebuilds at the length it declared (:issue:`1422`). - - A section header keeps the byte order its Byte-Order Magic declares. - One of 12 octets has no room for the magic, and takes the byte order - in which its Block Total Length reads as 12 (see - :class:`~pcapkit.protocols.schema.misc.pcapng.SectionHeaderBlock`). - - """ - if isinstance(schema, Schema_UnknownBlock): - body = schema.body - else: - # the octets unpacked for the fixed fields, which start with the - # Block Total Length and run past the end of the block - octets = bytes(schema) - body = octets[4:schema.length - 8] - warn(f'PCAP-NG: [Block {header.type}] block length {schema.length} is below the ' - f'{self.SHORT_HEADER_TYPES[header.type]}-octet minimum; kept as captured', - ProtocolWarning, stacklevel=stacklevel()) - if octets[schema.length - 8:schema.length - 4] != octets[:4]: - warn(f'PCAP-NG: [Block {header.type}] block length mismatch: {schema.length} != ' - f'{octets[schema.length - 8:schema.length - 4].hex()}', ProtocolWarning, - stacklevel=stacklevel()) - - data: 'Data_SectionHeaderBlock | Data_InterfaceDescriptionBlock' - if header.type == Enum_BlockType.Section_Header_Block: - # NOTE: On the construction path :meth:`_make_block_shb` has already - # set the byte order from the data model. - if isinstance(schema, Schema_SectionHeaderBlock): - self._byte = schema.byteorder - endian = '<' if self._byte == 'little' else '>' - major, minor, section_length = struct.unpack(f'{endian}HHq', (body + bytes(16))[4:16]) - data = Data_SectionHeaderBlock( - type=header.type, - length=schema.length, - byteorder=self._byte, - version=VersionInfo( - major=major, - minor=minor, - ), - section_length=section_length, - options=self._read_pcapng_options([]), - ) - else: - endian = '<' if self._byte == 'little' else '>' - linktype, _, snaplen = struct.unpack(f'{endian}HHI', (body + bytes(8))[:8]) - data = Data_InterfaceDescriptionBlock( - type=header.type, - length=schema.length, - linktype=Enum_LinkType.get(linktype), - snaplen=snaplen, - options=self._read_pcapng_options([]), - ) - data.__update__([('block_raw', body)]) - return data - def _read_pcapng_options(self, options_schema: 'list[Schema_Option]') -> 'Option': """Read PCAP-NG options. @@ -4737,22 +4753,6 @@ def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *, length2=packet_len + BLOCK_MINIMUM[Enum_BlockType.Simple_Packet_Block], ), block, 'padding', 'padding_data') - def _make_block_short( - self, block: 'Data_EnhancedPacketBlock | Data_PacketBlock | Data_SectionHeaderBlock | Data_InterfaceDescriptionBlock', # pylint: disable=line-too-long - ) -> 'Schema_UnknownBlock': - """Make a PCAP-NG EPB, packet block (obsolete), SHB or IDB kept as captured. - - Args: - block: Block data model, whose ``block_raw`` holds the body of a - block shorter than its fixed fields (see :meth:`_read_block_short` - and :meth:`_read_block_short_header`). - - Returns: - Constructed block schema, the body written back verbatim. - - """ - return self._make_block_unknown(data=block.block_raw) # type: ignore[union-attr] - def _make_block_nrb(self, block: 'Optional[Data_NameResolutionBlock]' = None, *, records: 'Optional[Record | list[Schema_NameResolutionRecord | tuple[Enum_RecordType, dict[str, Any]] | bytes]]' = None, options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None, diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 725a8b73fd..6afba3ba4d 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -840,31 +840,6 @@ def pack(self, **kwargs: 'Any') -> 'bytes': packet = kwargs.get('__packet__', {}) # packet data return header.pack(packet) - def _packing_target(self) -> 'Self': - """Return the instance that :meth:`pack` makes its header on. - - Returns: - The instance itself while it is being constructed, i.e. before - :meth:`__post_init__` has set its :attr:`info`, since its - :meth:`unpack` reads the made header back; a shallow copy of it - otherwise, so that neither that header nor an attribute that - :meth:`make` sets is left on the instance (:issue:`1769`). - - Note: - A :meth:`pack` override that does not call this one has to build - on this target too, as :meth:`Frame.pack - ` and - :meth:`PCAPNG.pack ` do. - The copy is shallow, so a :meth:`make` that mutates a container - in place still reaches the instance's. - - """ - # NOTE: ``_info`` is the last thing ``__post_init__`` sets, from - # ``unpack``; until then the instance holds no packet to keep. - if '_info' not in vars(self): - return self - return copy.copy(self) - def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT': """Unpack (parse) packet data. @@ -1088,56 +1063,6 @@ class object, or a :class:`~pcapkit.corekit.module.ModuleDescriptor` f'{incumbent!r} with {entry!r}', RegistryWarning) cls.__proto__[code] = entry - @classmethod - def _next_layer_entry(cls, code: 'Any', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]' - ) -> 'Optional[ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]': - """Work out what a next-layer ``register`` stores under ``code``. - - :meth:`register` and every override of it, such as - :meth:`Link.register `, - share this; each still warns and stores for itself. - - Arguments: - code: registry key - protocol: module descriptor or a :class:`ProtocolBase` subclass - - Returns: - :obj:`None` if ``protocol`` is the entry stored under ``code`` - already -- the same object, or a descriptor equal to a stored one - (:issue:`1364`). Otherwise the entry to store: if ``protocol`` is a - descriptor equal to the one ``code`` shipped with, per - :attr:`_shipped_proto`, the shipped object itself, unresolved; - else ``protocol`` resolved to its class. - - Raises: - pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a - class, or not a :class:`ProtocolBase` subclass. - - Note: - The shipped descriptor is matched before it is resolved, so an - override undone by handing back the shipped entry leaves the - registry as it shipped: the entry is still a descriptor, nothing is - imported, and the class is looked up when a frame dispatches to it - (:issue:`1504`). - - """ - incumbent = cls.__proto__.get(code) - if incumbent is not None and (incumbent is protocol or ( - isinstance(incumbent, ModuleDescriptor) and isinstance(protocol, ModuleDescriptor) - and incumbent == protocol)): - return None - shipped = cls._shipped_proto.get(code) - if (isinstance(protocol, ModuleDescriptor) and isinstance(shipped, ModuleDescriptor) - and protocol == shipped): - return shipped - if isinstance(protocol, ModuleDescriptor): - protocol = protocol.klass - if not isinstance(protocol, type): - raise RegistryError(f'protocol must be a class, not {protocol!r}') - if not issubclass(protocol, ProtocolBase): - raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') - return protocol - @classmethod def from_schema(cls, schema: '_ST | dict[str, Any]') -> 'Self': """Create protocol instance from schema. @@ -1731,6 +1656,94 @@ def __hash__(self) -> 'int': # Utilities. ########################################################################## + @classmethod + def _make_data(cls, data: 'Data') -> 'dict[str, Any]': + """Create key-value pairs from ``data`` for protocol construction. + + Args: + data: protocol data + + Returns: + Key-value pairs for protocol construction. + + """ + return data.to_dict() + + def _packing_target(self) -> 'Self': + """Return the instance that :meth:`pack` makes its header on. + + Returns: + The instance itself while it is being constructed, i.e. before + :meth:`__post_init__` has set its :attr:`info`, since its + :meth:`unpack` reads the made header back; a shallow copy of it + otherwise, so that neither that header nor an attribute that + :meth:`make` sets is left on the instance (:issue:`1769`). + + Note: + A :meth:`pack` override that does not call this one has to build + on this target too, as :meth:`Frame.pack + ` and + :meth:`PCAPNG.pack ` do. + The copy is shallow, so a :meth:`make` that mutates a container + in place still reaches the instance's. + + """ + # NOTE: ``_info`` is the last thing ``__post_init__`` sets, from + # ``unpack``; until then the instance holds no packet to keep. + if '_info' not in vars(self): + return self + return copy.copy(self) + + @classmethod + def _next_layer_entry(cls, code: 'Any', protocol: 'ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]' + ) -> 'Optional[ModuleDescriptor[ProtocolBase] | Type[ProtocolBase]]': + """Work out what a next-layer ``register`` stores under ``code``. + + :meth:`register` and every override of it, such as + :meth:`Link.register `, + share this; each still warns and stores for itself. + + Arguments: + code: registry key + protocol: module descriptor or a :class:`ProtocolBase` subclass + + Returns: + :obj:`None` if ``protocol`` is the entry stored under ``code`` + already -- the same object, or a descriptor equal to a stored one + (:issue:`1364`). Otherwise the entry to store: if ``protocol`` is a + descriptor equal to the one ``code`` shipped with, per + :attr:`_shipped_proto`, the shipped object itself, unresolved; + else ``protocol`` resolved to its class. + + Raises: + pcapkit.utilities.exceptions.RegistryError: If ``protocol`` is not a + class, or not a :class:`ProtocolBase` subclass. + + Note: + The shipped descriptor is matched before it is resolved, so an + override undone by handing back the shipped entry leaves the + registry as it shipped: the entry is still a descriptor, nothing is + imported, and the class is looked up when a frame dispatches to it + (:issue:`1504`). + + """ + incumbent = cls.__proto__.get(code) + if incumbent is not None and (incumbent is protocol or ( + isinstance(incumbent, ModuleDescriptor) and isinstance(protocol, ModuleDescriptor) + and incumbent == protocol)): + return None + shipped = cls._shipped_proto.get(code) + if (isinstance(protocol, ModuleDescriptor) and isinstance(shipped, ModuleDescriptor) + and protocol == shipped): + return shipped + if isinstance(protocol, ModuleDescriptor): + protocol = protocol.klass + if not isinstance(protocol, type): + raise RegistryError(f'protocol must be a class, not {protocol!r}') + if not issubclass(protocol, ProtocolBase): + raise RegistryError(f'protocol must be a Protocol subclass, not {protocol!r}') + return protocol + def _get_context(self, cls: 'Optional[Type[_CTX]]' = None) -> 'Optional[_CTX]': """Get the caller supplied context for this protocol, if any. @@ -2159,19 +2172,6 @@ def _make_index_namespace(cls, name: 'str', f'{", ".join(found)} ({matches}); pass {hint} explicitly') return next(iter(found.values())) - @classmethod - def _make_data(cls, data: 'Data') -> 'dict[str, Any]': - """Create key-value pairs from ``data`` for protocol construction. - - Args: - data: protocol data - - Returns: - Key-value pairs for protocol construction. - - """ - return data.to_dict() - @classmethod def _make_payload(cls, data: 'Data') -> 'ProtocolBase': """Create payload from ``data`` for protocol construction. diff --git a/pcapkit/protocols/schema/internet/ipv6.py b/pcapkit/protocols/schema/internet/ipv6.py index da65275288..7b16716a38 100644 --- a/pcapkit/protocols/schema/internet/ipv6.py +++ b/pcapkit/protocols/schema/internet/ipv6.py @@ -88,13 +88,6 @@ class IPv6(Schema): #: padding, see :attr:`IPv4.trailer `. trailer: 'bytes' = PayloadField(default=b'') - if TYPE_CHECKING: - def __init__(self, hextet: 'IPv6Hextet', length: 'int', next: 'Enum_TransType', - limit: 'int', src: 'IPv6Address | bytes | str | int', - dst: 'IPv6Address | bytes | str | int', - payload: 'bytes | ProtocolBase | Schema', - trailer: 'bytes' = b'') -> None: ... - def post_process(self, packet: 'dict[str, Any]') -> 'Schema': """Revise ``schema`` data after unpacking process. @@ -125,6 +118,13 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': self.trailer = self.__buffer__['trailer'] = rest[jumbo:] return self + if TYPE_CHECKING: + def __init__(self, hextet: 'IPv6Hextet', length: 'int', next: 'Enum_TransType', + limit: 'int', src: 'IPv6Address | bytes | str | int', + dst: 'IPv6Address | bytes | str | int', + payload: 'bytes | ProtocolBase | Schema', + trailer: 'bytes' = b'') -> None: ... + def jumbo_payload_length(next: 'int', data: 'bytes') -> 'Optional[int]': # pylint: disable=redefined-builtin """Find the Jumbo Payload Length of an IPv6 packet [:rfc:`2675`]. diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 07b689e649..71d717f397 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -821,12 +821,6 @@ class FlagsField(UInt32Field): """ - def __init__(self, namespace: 'dict[str, tuple[int, int]]', - callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None) -> 'None': - super().__init__(callback=callback) - - self._flags = namespace - def pre_process(self, value: 'dict[str, int]', packet: 'dict[str, Any]') -> 'int | bytes': # type: ignore[override] """Process field value before construction (packing). @@ -864,6 +858,12 @@ def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'dict[ word = super().post_process(value, packet) return {name: word >> start & ((1 << size) - 1) for name, (start, size) in self._flags.items()} + def __init__(self, namespace: 'dict[str, tuple[int, int]]', + callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None) -> 'None': + super().__init__(callback=callback) + + self._flags = namespace + #: Sub-fields of the ``epb_flags`` and ``pack_flags`` words. ``reserved`` keeps #: bits 12 to 23 verbatim -- the reserved bits 12-15 and the link-layer-dependent @@ -909,16 +909,6 @@ class OptionEnumField(EnumField): if TYPE_CHECKING: _namespace: 'Enum_OptionType' - def __init__(self, length: 'int | Callable[[dict[str, Any]], int]', - default: 'Enum_OptionType' = Enum_OptionType.opt_endofopt, signed: 'bool' = False, - byteorder: 'Literal["little", "big"]' = 'big', - bit_length: 'Optional[int]' = None, - namespace: 'str' = 'opt', - callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None) -> 'None': - super().__init__(length, default, signed, byteorder, bit_length, Enum_OptionType, callback) - - self._opt_ns = namespace - def pre_process(self, value: 'int | Enum_OptionType', packet: 'dict[str, Any]') -> 'int | bytes': """Process field value before construction (packing). @@ -1002,6 +992,16 @@ def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'Enum_ opt_name=opt_name, opt_value=value) return self._namespace.get(value, namespace=namespace) + def __init__(self, length: 'int | Callable[[dict[str, Any]], int]', + default: 'Enum_OptionType' = Enum_OptionType.opt_endofopt, signed: 'bool' = False, + byteorder: 'Literal["little", "big"]' = 'big', + bit_length: 'Optional[int]' = None, + namespace: 'str' = 'opt', + callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None) -> 'None': + super().__init__(length, default, signed, byteorder, bit_length, Enum_OptionType, callback) + + self._opt_ns = namespace + @schema_final class PCAPNG(Schema): @@ -1084,68 +1084,6 @@ class Option(EnumSchema[Enum_OptionType]): }, ) - def __init_subclass__(cls, /, code: 'Optional[Enum_OptionType | Iterable[Enum_OptionType]]' = None, - ns: 'Optional[str]' = None, *args: 'Any', **kwargs: 'Any') -> 'None': - """Register option type to :attr:`__enum__` mapping. - - Args: - code: Option type code. It can be either a single option type enumeration - or a list of option type enumerations. - ns: Namespace of option type enumeration. If not given, the value - will be inferred from the option type code. Spelled ``ns`` rather - than ``namespace`` because ``abc.ABCMeta.__new__()`` names its - own fourth parameter ``namespace``, and before Python 3.11 that - parameter is positional-or-keyword rather than positional-only -- - so a class keyword literally called ``namespace`` binds it twice. - *args: Arbitrary positional arguments. - **kwargs: Arbitrary keyword arguments. - - If ``code`` is provided, the subclass will be registered to the - :attr:`__enum__` mapping with the given ``code``. If ``code`` is - not given, the subclass will not be registered. - - Examples: - - .. code-block:: python - - from pcapkit.const.pcapng.option_type import OptionType as Enum_OptionType - from pcapkit.protocols.schema.misc.pcapng import Option - - class NewOption(Option, ns='opt', code=Enum_OptionType.opt_new): - ... - - See Also: - - :class:`pcapkit.const.pcapng.option_type.OptionType` - - .. versionchanged:: 1.5.0 - - The class keyword ``namespace`` is renamed ``ns``, with no alias - (:issue:`439`). - - """ - # NOTE: the base hook goes first, before any of the registration work - # below -- see :meth:`EnumSchema.__init_subclass__`, which makes the same - # move for the same reason. :meth:`Schema.__init_subclass__` is what - # refuses to derive from a finalised schema, and a refusal has to land - # before :meth:`Option.register` writes ``cls`` into ``__enum__``: calling - # this last would let a raise discard the class object while leaving the - # registry pointing at it, so a rejected declaration would still - # displace a built-in option schema for the rest of the process. - super().__init_subclass__() - - if ns is not None: - cls.__namespace__ = ns - - if code is not None: - if ns is None: - ns = cast('Optional[str]', cls.__namespace__) - - if not isinstance(code, Enum_OptionType): - for _code in code: - Option.register(_code, cls, ns) - else: - Option.register(code, cls, ns) - @staticmethod def register(code: 'Enum_OptionType', cls: 'Type[Option]', ns: 'Optional[str]' = None) -> 'None': """Register option type to :attr:`__enum__` mapping. @@ -1210,6 +1148,68 @@ def register(code: 'Enum_OptionType', cls: 'Type[Option]', ns: 'Optional[str]' = for key in targets: Option.registry[key][code] = cls + def __init_subclass__(cls, /, code: 'Optional[Enum_OptionType | Iterable[Enum_OptionType]]' = None, + ns: 'Optional[str]' = None, *args: 'Any', **kwargs: 'Any') -> 'None': + """Register option type to :attr:`__enum__` mapping. + + Args: + code: Option type code. It can be either a single option type enumeration + or a list of option type enumerations. + ns: Namespace of option type enumeration. If not given, the value + will be inferred from the option type code. Spelled ``ns`` rather + than ``namespace`` because ``abc.ABCMeta.__new__()`` names its + own fourth parameter ``namespace``, and before Python 3.11 that + parameter is positional-or-keyword rather than positional-only -- + so a class keyword literally called ``namespace`` binds it twice. + *args: Arbitrary positional arguments. + **kwargs: Arbitrary keyword arguments. + + If ``code`` is provided, the subclass will be registered to the + :attr:`__enum__` mapping with the given ``code``. If ``code`` is + not given, the subclass will not be registered. + + Examples: + + .. code-block:: python + + from pcapkit.const.pcapng.option_type import OptionType as Enum_OptionType + from pcapkit.protocols.schema.misc.pcapng import Option + + class NewOption(Option, ns='opt', code=Enum_OptionType.opt_new): + ... + + See Also: + - :class:`pcapkit.const.pcapng.option_type.OptionType` + + .. versionchanged:: 1.5.0 + + The class keyword ``namespace`` is renamed ``ns``, with no alias + (:issue:`439`). + + """ + # NOTE: the base hook goes first, before any of the registration work + # below -- see :meth:`EnumSchema.__init_subclass__`, which makes the same + # move for the same reason. :meth:`Schema.__init_subclass__` is what + # refuses to derive from a finalised schema, and a refusal has to land + # before :meth:`Option.register` writes ``cls`` into ``__enum__``: calling + # this last would let a raise discard the class object while leaving the + # registry pointing at it, so a rejected declaration would still + # displace a built-in option schema for the rest of the process. + super().__init_subclass__() + + if ns is not None: + cls.__namespace__ = ns + + if code is not None: + if ns is None: + ns = cast('Optional[str]', cls.__namespace__) + + if not isinstance(code, Enum_OptionType): + for _code in code: + Option.register(_code, cls, ns) + else: + Option.register(code, cls, ns) + if TYPE_CHECKING: #: Option type. type: 'Enum_OptionType' @@ -2311,6 +2311,37 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Self': self.data = [self.entry] return self + @staticmethod + def dump_entries(entries: 'list[OrderedMultiDict[str, str | bytes]]') -> 'bytes': + """Serialise journal entries, without the block's 32-bit padding. + + Args: + entries: The journal entries. + + Returns: + Each field as ``KEY=value\\n``, or for a binary value as the name, + a newline, its 64-bit little-endian length, the value and a newline, + with one blank line between entries. + + """ + temp = [] # type: list[bytes] + for entry in entries: + tmp_buf = [] # type: list[bytes] + for key, val in entry.items(multi=True): + if isinstance(val, str): + tmp_buf.append(f'{key}={val}\n'.encode()) + else: + tmp_buf.append(b'%s\n%s%s\n' % (key.encode(), struct.pack(' 'None': ... + def _read_entries(self) -> 'list[OrderedMultiDict[str, str | bytes]]': """Walk :attr:`entry` into its journal entries, as :meth:`post_process` describes. @@ -2372,30 +2403,6 @@ def _read_entries(self) -> 'list[OrderedMultiDict[str, str | bytes]]': break return data - @staticmethod - def dump_entries(entries: 'list[OrderedMultiDict[str, str | bytes]]') -> 'bytes': - """Serialise journal entries, without the block's 32-bit padding. - - Args: - entries: The journal entries. - - Returns: - Each field as ``KEY=value\\n``, or for a binary value as the name, - a newline, its 64-bit little-endian length, the value and a newline, - with one blank line between entries. - - """ - temp = [] # type: list[bytes] - for entry in entries: - tmp_buf = [] # type: list[bytes] - for key, val in entry.items(multi=True): - if isinstance(val, str): - tmp_buf.append(f'{key}={val}\n'.encode()) - else: - tmp_buf.append(b'%s\n%s%s\n' % (key.encode(), struct.pack(' 'str': """Decode a journal field name, key or value, reporting what did not decode. @@ -2417,13 +2424,6 @@ def _decode_text(octets: 'bytes') -> 'str': f'decode', SchemaWarning, stacklevel=stacklevel()) return octets.decode('utf-8', errors='replace') - if TYPE_CHECKING: - #: Journal entry (decoded), or the entry as captured, as one - #: :obj:`bytes` item, if it is malformed. - data: 'list[OrderedMultiDict[str, str | bytes] | bytes]' - - def __init__(self, length: 'int', entry: 'bytes', length2: 'int') -> 'None': ... - class DSBSecrets(EnumSchema[Enum_SecretsType]): """Header schema for DSB secrets data.""" diff --git a/pcapkit/protocols/schema/schema.py b/pcapkit/protocols/schema/schema.py index 6d79c11507..115f8cc507 100644 --- a/pcapkit/protocols/schema/schema.py +++ b/pcapkit/protocols/schema/schema.py @@ -416,417 +416,124 @@ class Schema(Mapping[str, _VT], Generic[_VT], metaclass=SchemaMeta): #: List of names to be excluded from :obj:`dict` conversion. __excluded__: 'list[str]' = [] - def __init_subclass__(cls, /, *args: 'Any', **kwargs: 'Any') -> 'None': - """Refuse to derive from a finalised schema. + @classmethod + def from_dict(cls, dict_: 'Optional[Mapping[str, _VT] | Iterable[tuple[str, _VT]]]' = None, + **kwargs: '_VT') -> 'Self': + r"""Create a new instance. - :func:`~pcapkit.utilities.compat.final` is a promise to the type - checker and nothing more: it records ``__final__`` on the class and - leaves the interpreter free to subclass it anyway. Every schema - :func:`schema_final` finalises carries a generated ``__init__`` built - from the :attr:`__fields__` that were declared at that moment, so a - subclass adding a field afterwards inherits a constructor that cannot - set it. This turns that promise into a rule the interpreter keeps. + * If ``dict_`` is a mapping, then does: + ``for k in dict_: self[k] = dict_[k]``. + * If ``dict_`` is present and not a mapping, then does: + ``for k, v in dict_: self[k] = v``. + * In either case, and if ``dict_`` is not present, this is followed + by: ``for k, v in kwargs.items(): self[k] = v``. Args: - *args: Arbitrary positional arguments. - **kwargs: Arbitrary keyword arguments in class definition. - - Raises: - SchemaError: If any class in ``cls``'s ancestry carries the - ``__final__`` marker in its own ``__dict__``. + dict\_: Source data. + **kwargs: Arbitrary keyword arguments. """ - # NOTE: the whole ancestry rather than ``cls.__bases__``, and - # ``base.__dict__`` rather than ``getattr`` -- see - # :meth:`pcapkit.corekit.infoclass.Info.__init_subclass__`, which makes - # the same choices for the same reasons. - for base in cls.__mro__[1:]: - if base.__dict__.get('__final__'): - raise SchemaError(f'{cls.__name__}: cannot subclass {base.__name__}, ' - 'which is final') - - # NOTE: forwarded rather than swallowed, so that a class keyword nobody - # accepts still reaches ``object`` and fails there. - # :meth:`EnumSchema.__init_subclass__` is the one caller that reaches here - # with nothing: it consumes its own ``code`` keyword and discards the - # rest, which is why a stray class keyword is tolerated on an - # :class:`EnumSchema` subclass and rejected on a plain one. - super().__init_subclass__(*args, **kwargs) + self = cls.__new__(cls) + self.__update__(dict_, **kwargs) + self.__post_init__() + return self - def __new__(cls, *args: '_VT', **kwargs: '_VT') -> 'Self': # pylint: disable=unused-argument - """Create a new instance. + def to_dict(self) -> 'dict[str, _VT]': + """Convert :class:`Schema` into :obj:`dict`. - The class will try to automatically generate ``__init__`` method with - one keyword parameter per field in :attr:`__fields__`, which is - inspired by :pep:`557` (:mod:`dataclasses`). + Important: + We only convert nested :class:`Schema` objects into :obj:`dict` if + they are the direct value of the :class:`Schema` object's attribute. + Should such :class:`Schema` objects be nested within other data + types, such as :obj:`list`, :obj:`tuple`, :obj:`set`, etc., we + leave them intact rather than converting them into :obj:`dict`. Note: - Whether ``cls`` still needs finalising is read from its *own* - ``__dict__``, never inherited (:issue:`1498`). Each - class is finalised on its first construction, whatever its - ancestors' state, so a subclass of a - :attr:`~pcapkit.corekit.infoclass.FinalisedState.BASE` ancestor - still gets its own ``__builtin__``, ``__excluded__`` and generated - ``__init__``, and a ``@final`` class under a ``BASE`` ancestor is - refused like any other. - - Args: - *args: Arbitrary positional arguments. - **kwargs: Arbitrary keyword arguments. + A schema unpacked as the value of a + :class:`~pcapkit.corekit.fields.misc.SchemaField` that left part of + the field's span unread carries those octets under a + ``__remainder__`` key, and the number of octets it did read under + ``__remainder_offset__``. :meth:`from_dict` accepts both back, so + that the rebuilt field still packs them (:issue:`1380`). The keys + are absent otherwise. - Raises: - SchemaError: If ``cls`` was marked with - :func:`~pcapkit.utilities.compat.final` but never finalised by - :func:`schema_final`, i.e. it carries ``__final__`` in its own - ``__dict__`` without - :attr:`~pcapkit.corekit.infoclass.FinalisedState.FINAL`. + A schema whose data ended inside one of its fields carries the + ``__short_read__`` that :meth:`unpack` recorded, and + :meth:`from_dict` accepts it back, likewise. """ - # NOTE: first instantiation is the earliest point a bare ``@final`` can - # be caught -- ``final`` runs after the class object exists, so - # :meth:`__init_subclass__` has already returned. Nested inside the NONE - # branch, which is already one-shot per class, so a finalised schema pays - # nothing for it. See :meth:`pcapkit.corekit.infoclass.Info.__new__`, - # which carries the full reasoning; on this side the missing generated - # ``__init__`` is the one built from :attr:`__fields__`. - # - # Both lookups read ``cls.__dict__`` rather than ``getattr``, because - # both markers are ordinary, inherited class attributes. The outer one: - # :func:`schema_final` writes ``BASE`` onto the class it finalises, and a - # subclass reading that state through inheritance was never finalised - # itself -- it would keep the ``__excluded__`` it copied at declaration, - # without the names :func:`schema_final` adds, and leak ``__map__``, - # ``__map_reverse__``, ``__buffer__`` and ``__updated__`` from - # :meth:`to_dict` (GitHub issue #1498). The inner one: an ancestor - # hand-marked bare ``@final`` passes its marker to every subclass - # declared before it landed, and ``getattr`` would blame such a subclass - # for a mismarking that is its ancestor's. - if cls.__dict__.get('__finalised__', FinalisedState.NONE) == FinalisedState.NONE: - if cls.__dict__.get('__final__'): - raise SchemaError(f'{cls.__name__}: marked final but never finalised, so it has no generated ' - '__init__; apply schema_final, which applies final itself, not final alone') - cls = schema_final(cls, _finalised=False) - self = super().__new__(cls) + dict_ = {} # type: dict[str, Any] + for (key, value) in self.__dict__.items(): + if key in self.__excluded__: + continue - # NOTE: We define the ``__map__`` and ``__map_reverse__`` attributes - # here under ``self`` to avoid them being considered as class variables - # and thus being shared by all instances. - super().__setattr__(self, '__map__', {}) - super().__setattr__(self, '__map_reverse__', {}) + out_key = self.__map_reverse__.get(key, key) + if isinstance(value, Schema): + dict_[out_key] = value.to_dict() + else: + dict_[out_key] = value - # NOTE: We only create the attributes for the instance itself, - # to avoid creating shared attributes for the class. - super().__setattr__(self, '__buffer__', {name: b'' for name in self.__fields__.keys()}) - super().__setattr__(self, '__updated__', True) + for key in ('__remainder__', '__remainder_offset__', '__short_read__'): + if key in self.__dict__: + dict_[key] = self.__dict__[key] + return dict_ - return self + def to_bytes(self) -> 'bytes': + """Convert :class:`Schema` into :obj:`bytes`.""" + return self.__bytes__() - def __post_init__(self, packet: 'Optional[dict[str, Any]]' = None) -> 'None': - """Fill in the fields the caller left unset. + def get_payload(self, name: 'Optional[str]' = None) -> 'bytes': + """Get payload of :class:`Schema`. Args: - packet: Packet data, as forwarded from the ``__packet__`` keyword - argument of the generated ``__init__``. The schema is packed - here only when one is given; see the note below. + name: Name of the payload field. + + Returns: + Payload of :class:`Schema` as :obj:`bytes`. """ - for name, field in self.__fields__.items(): - # NOTE: Read with a fallback rather than by subscript, since the - # generated ``__init__`` is not the only caller: :meth:`from_dict` - # seeds only the keys its argument carries, so a field the caller left - # out is missing from ``__dict__`` entirely rather than holding - # ``NO_VALUE``. - # - # What is tested is ``NO_VALUE`` alone, not ``NO_VALUE`` or ``None``. - # This method fills in what the caller did not say, and a ``None`` the - # caller passed *is* something said: on an optional field it is the - # chosen value, meaning this packet does not carry the field. It is - # also what :meth:`unpack` stores for a - # :class:`~pcapkit.corekit.fields.misc.ConditionalField` whose test - # fails -- including one that declares a default of its own -- so - # substituting the default here would leave a constructed schema - # disagreeing with a parsed one about the same packet, and - # ``from_dict(parsed.to_dict())`` no longer reproducing what it was - # given. Telling the two apart is what ``NO_VALUE`` is for. - value = self.__dict__.get(name, NO_VALUE) - if value is not NO_VALUE: - continue + if name is None: + name = self.__payload__ - default = field.default - if default is not NO_VALUE: - self.__dict__[name] = default - else: - # NOTE: Nothing to fill an unset field with, so the ``NO_VALUE`` - # the generated ``__init__`` seeded it with is dropped rather than - # kept: it is a *field* sentinel, not a value a schema may hold. - # Dropping it rather than storing ``None`` also keeps the name out - # of the context :meth:`pack` builds from ``__dict__``, which a - # schema may be relying on to seed for itself -- the PCAP-NG - # section header block reads its Byte-Order Magic from a ``match`` - # its own :meth:`pre_pack` supplies, and only when the context - # does not name one already. :meth:`pack` reads an absent field as - # ``None`` regardless. - self.__dict__.pop(name, None) + field = self.__fields__.get(name) + if field is None: + raise ProtocolUnbound(f'unknown field: {name!r}') + if not isinstance(field, PayloadField): + raise ProtocolUnbound(f'not a payload field: {name!r}') + return self.__buffer__[name] - # NOTE: Packed here only when a packet context was actually handed over. - # A schema is not in general packable from its own fields alone: a field - # callback may read a key that the *enclosing* layer owns, as the - # Multipath TCP options do with the ``length`` of the TCP option that - # carries them, and packing without it raises rather than producing - # octets. ``__updated__`` is still set, so a schema left unpacked here is - # packed by :meth:`__bytes__` on first use -- by which time the enclosing - # layer has supplied the context. - if packet is not None: - self.pack(packet) + def pack(self, packet: 'Optional[dict[str, Any]]' = None) -> 'bytes': + """Pack :class:`Schema` into :obj:`bytes`. - def __update__(self, dict_: 'Optional[Mapping[str, _VT] | Iterable[tuple[str, _VT]]]' = None, - **kwargs: '_VT') -> 'None': - # NOTE: Keys with the same names as the class's builtin methods will be - # renamed with the class name prefixed as mangled class variables - # implicitly and internally. Such mapping information will be stored - # within the ``__map__`` attribute. + Args: + packet: Packet data. - __name__ = type(self).__name__ # pylint: disable=redefined-builtin + Returns: + Packed :class:`Schema` as :obj:`bytes`. - if dict_ is None: - data_iter = kwargs.items() # type: Iterable[tuple[str, Any]] - elif isinstance(dict_, collections.abc.Mapping): - data_iter = itertools.chain(dict_.items(), kwargs.items()) - else: - data_iter = itertools.chain(dict_, kwargs.items()) + Notes: + Since we do not know the length of the packet, we use a + reasonable default value ``-1`` for the ``__length__`` + field, as the :class:`~pcapkit.corekit.fields.field.Field` + class will consider negative value as a placeholder. - for (key, value) in data_iter: - # NOTE: Not a field, but what :meth:`to_dict` emits for the octets an - # enclosing ``SchemaField`` span held beyond this schema, and where, - # so that a rebuilt schema still re-emits them (#1380). - # NOTE: Nor is the short read :meth:`unpack` records (#1458). - if key in ('__remainder__', '__remainder_offset__', '__short_read__'): - self.__dict__[key] = value - continue + If you want to pack the packet with the correct length, + please provide the ``__length__`` value before packing. - if key not in self.__buffer__: - warn(f'{key!r} is not a valid field name', UnknownFieldWarning) - continue + """ + if packet is None: + packet = {} - if key in self.__builtin__: - new_key = f'_{__name__}{key}' - - # NOTE: We keep record of the mapping bidirectionally. - self.__map__[key] = new_key - self.__map_reverse__[new_key] = key - - key = new_key - - # if key in self.__dict__: - # raise KeyExists(f'{key!r} already exists') - - # NOTE: We don't rewrite the key names here, just keep the - # original ones, even though they might break on the ``.`` - # (:obj:`getattr`) operator. - - # if isinstance(key, str): - # key = re.sub(r'\W', '_', key) - self.__dict__[key] = value - - self.__updated__ = True - - __init__ = __update__ - - def __str__(self) -> 'str': - temp = [] # type: list[str] - for (key, value) in self.__dict__.items(): - if key in self.__excluded__: - continue - - out_key = self.__map_reverse__.get(key, key) - temp.append(f'{out_key}={value}') - args = ', '.join(temp) - return f'{type(self).__name__}({args})' - - def __repr__(self) -> 'str': - temp = [] # type: list[str] - for (key, value) in self.__dict__.items(): - if key in self.__excluded__: - continue - - out_key = self.__map_reverse__.get(key, key) - if isinstance(value, Schema): - temp.append(f'{out_key}={type(value).__name__}(...)') - else: - temp.append(f'{out_key}={value!r}') - args = ', '.join(temp) - return f'{type(self).__name__}({args})' - - def __bytes__(self) -> 'bytes': - if self.__updated__: - self.pack() - - buffer = [] # type: list[bytes] - for name in self.__fields__.keys(): - value = self.__buffer__[name] - buffer.append(value) - return b''.join(buffer) - - def __len__(self) -> 'int': - return len(self.__bytes__()) - - def __iter__(self) -> 'Iterator[str]': - for key in self.__dict__: - if key in self.__builtin__: - continue - yield self.__map_reverse__.get(key, key) - - def __getitem__(self, name: 'str') -> '_VT': - # NOTE: Serves every key :meth:`__iter__` yields, not only the declared - # fields: a ``post_process`` may store derived values on the instance, - # as :class:`~pcapkit.protocols.schema.internet.ipv4.TSOption` does, and - # iteration reports them, so ``dict(schema)`` must be able to read them - # back (#1291). A mangled key is reachable only under the name - # it was mangled from, as in iteration. - if name in self.__fields__ or (name not in self.__builtin__ and name not in self.__map_reverse__): - key = self.__map__.get(name, name) - return self.__dict__[key] - raise KeyError(name) - - def __setattr__(self, name: 'str', value: '_VT') -> 'None': - if name in self.__fields__: - key = self.__map__.get(name, name) - self.__dict__[key] = value - # NOTE: ``self.__updated__ = True`` would re-enter this method once - # per field assigned, only to miss the ``__fields__`` test and fall - # through to ``object.__setattr__``. ``__updated__`` is an instance - # attribute established in ``__new__``, so the direct store is the same - # write with none of the round trip. - self.__dict__['__updated__'] = True - return - return super().__setattr__(name, value) - - def __delattr__(self, name: 'str') -> 'None': - if name in self.__fields__: - key = self.__map__.get(name, name) - del self.__dict__[key] - self.__updated__ = True - return - return super().__delattr__(name) - - @classmethod - def from_dict(cls, dict_: 'Optional[Mapping[str, _VT] | Iterable[tuple[str, _VT]]]' = None, - **kwargs: '_VT') -> 'Self': - r"""Create a new instance. - - * If ``dict_`` is a mapping, then does: - ``for k in dict_: self[k] = dict_[k]``. - * If ``dict_`` is present and not a mapping, then does: - ``for k, v in dict_: self[k] = v``. - * In either case, and if ``dict_`` is not present, this is followed - by: ``for k, v in kwargs.items(): self[k] = v``. - - Args: - dict\_: Source data. - **kwargs: Arbitrary keyword arguments. - - """ - self = cls.__new__(cls) - self.__update__(dict_, **kwargs) - self.__post_init__() - return self - - def to_dict(self) -> 'dict[str, _VT]': - """Convert :class:`Schema` into :obj:`dict`. - - Important: - We only convert nested :class:`Schema` objects into :obj:`dict` if - they are the direct value of the :class:`Schema` object's attribute. - Should such :class:`Schema` objects be nested within other data - types, such as :obj:`list`, :obj:`tuple`, :obj:`set`, etc., we - leave them intact rather than converting them into :obj:`dict`. - - Note: - A schema unpacked as the value of a - :class:`~pcapkit.corekit.fields.misc.SchemaField` that left part of - the field's span unread carries those octets under a - ``__remainder__`` key, and the number of octets it did read under - ``__remainder_offset__``. :meth:`from_dict` accepts both back, so - that the rebuilt field still packs them (:issue:`1380`). The keys - are absent otherwise. - - A schema whose data ended inside one of its fields carries the - ``__short_read__`` that :meth:`unpack` recorded, and - :meth:`from_dict` accepts it back, likewise. - - """ - dict_ = {} # type: dict[str, Any] - for (key, value) in self.__dict__.items(): - if key in self.__excluded__: - continue - - out_key = self.__map_reverse__.get(key, key) - if isinstance(value, Schema): - dict_[out_key] = value.to_dict() - else: - dict_[out_key] = value - - for key in ('__remainder__', '__remainder_offset__', '__short_read__'): - if key in self.__dict__: - dict_[key] = self.__dict__[key] - return dict_ - - def to_bytes(self) -> 'bytes': - """Convert :class:`Schema` into :obj:`bytes`.""" - return self.__bytes__() - - def get_payload(self, name: 'Optional[str]' = None) -> 'bytes': - """Get payload of :class:`Schema`. - - Args: - name: Name of the payload field. - - Returns: - Payload of :class:`Schema` as :obj:`bytes`. - - """ - if name is None: - name = self.__payload__ - - field = self.__fields__.get(name) - if field is None: - raise ProtocolUnbound(f'unknown field: {name!r}') - if not isinstance(field, PayloadField): - raise ProtocolUnbound(f'not a payload field: {name!r}') - return self.__buffer__[name] - - def pack(self, packet: 'Optional[dict[str, Any]]' = None) -> 'bytes': - """Pack :class:`Schema` into :obj:`bytes`. - - Args: - packet: Packet data. - - Returns: - Packed :class:`Schema` as :obj:`bytes`. - - Notes: - Since we do not know the length of the packet, we use a - reasonable default value ``-1`` for the ``__length__`` - field, as the :class:`~pcapkit.corekit.fields.field.Field` - class will consider negative value as a placeholder. - - If you want to pack the packet with the correct length, - please provide the ``__length__`` value before packing. - - """ - if packet is None: - packet = {} - - packet.update(self.__dict__) - # NOTE: ``pre_unpack`` is called here as well as from the unpacking path - # since a schema may seed ``packet`` keys that both paths need, as - # ``pcapkit.protocols.schema.internet.hip.EncryptedParameter`` does. - # ``pre_pack`` is what carries the packing-only preparation, such as the - # PCAP-NG section header block's byte order magic, which no field of the - # schema holds and which therefore cannot be recovered from - # ``self.__dict__`` above. - self.pre_unpack(packet) - self.pre_pack(packet) + packet.update(self.__dict__) + # NOTE: ``pre_unpack`` is called here as well as from the unpacking path + # since a schema may seed ``packet`` keys that both paths need, as + # ``pcapkit.protocols.schema.internet.hip.EncryptedParameter`` does. + # ``pre_pack`` is what carries the packing-only preparation, such as the + # PCAP-NG section header block's byte order magic, which no field of the + # schema holds and which therefore cannot be recovered from + # ``self.__dict__`` above. + self.pre_unpack(packet) + self.pre_pack(packet) if '__length__' not in packet: packet['__length__'] = -1 # reasonable default value @@ -965,43 +672,6 @@ class will consider negative value as a placeholder. self.__updated__ = False return self.__bytes__() - def _note_short_read(self, name: 'str', length: 'int', byte: 'bytes') -> 'None': - """Record the first field the data ends inside, for :meth:`_cut_short_read`. - - Args: - name: Field name. - length: Octets the field declares. - byte: Octets actually read for it. - - """ - if len(byte) < length and '__short_read__' not in self.__dict__: - self.__dict__['__short_read__'] = (name, len(byte)) - - def _cut_short_read(self) -> 'None': - """Cut the packed buffers back to the octets a short read consumed. - - A schema whose data ended inside one of its fields records, as - ``__short_read__``, that field's name and how many of its octets were - there (see :meth:`unpack`). Its value is read as though the missing - octets were zeros, so packing it writes the field at full width and - every field after it besides, none of which was captured. Only the - octets read are kept here, so a truncated header packs back to exactly - what was captured (:issue:`1458`). - - """ - record = self.__dict__.get('__short_read__') - if record is None: - return - - name, count = record - cut = False - for key in self.__fields__: - if cut: - self.__buffer__[key] = b'' - elif key == name: - self.__buffer__[key] = self.__buffer__[key][:count] - cut = True - def pre_pack(self, packet: 'dict[str, Any]') -> 'None': """Prepare ``packet`` data for packing process. @@ -1118,97 +788,427 @@ def unpack(cls, data: 'bytes | IO[bytes]', continue field = field.field(packet) - length = field.length + length = field.length + + # NOTE: bounded as the payload read above is (#1975). + byte = data.read(min(length, sys.maxsize)) + self.__buffer__[field.name] = byte + if not isinstance(field, ForwardMatchField): + self._note_short_read(field.name, length, byte) + + value = field.unpack(byte, packet.copy()) + if isinstance(field, SchemaField) and isinstance(value, Schema): + # NOTE: Kept on the nested schema rather than only in + # ``__buffer__``, so that the octets it left unread go wherever + # the value goes, :meth:`to_dict` included; see :meth:`pack`. + # The width read is summed from the nested buffers rather than + # taken with :func:`len`, which may pack the schema again. + consumed = sum(map(len, value.__buffer__.values())) + if len(byte) > consumed: + value.__dict__['__remainder__'] = byte[consumed:] + value.__dict__['__remainder_offset__'] = consumed + setattr(self, field.name, value) + + packet[field.name] = value + + if isinstance(field, OptionField): + packet['__option_padding__'] = field.option_padding + + if isinstance(field, ForwardMatchField): + # NOTE: a forward match reads ``length`` octets so a later field + # can size itself from them, but consumes neither the stream + # (the rewind below) nor ``__length__`` (no decrement in this + # branch). ``self.__buffer__[field.name]`` above still holds the + # octets just read, so it is zeroed here, as :meth:`pack` does for + # the same field type: ``__bytes__``/``__len__`` concatenate every + # slot in ``__buffer__``, and the octets are read again, for real, + # by whichever field actually needs them, so keeping them would + # over-report the schema's length by the forward match's width. + # Zeroing is also what lets a declared area checked against + # ``len(self)`` -- :class:`~pcapkit.corekit.fields.collections.OptionField` + # and :class:`~pcapkit.corekit.fields.collections.ListField` both + # do this -- see the octets actually consumed. + # + # The rewind is by the octets actually read, not by ``length``: + # near the end of the stream the read comes back short, and + # rewinding the full width would move the stream *before* the + # forward match, so the fields after it would re-read the + # preceding octet as their first. + data.seek(-len(byte), io.SEEK_CUR) + self.__buffer__[field.name] = b'' + elif isinstance(field, OptionField) and field.option_padding > 0: + # the option list ended before the declared field length was + # exhausted; give the unconsumed remainder back to ``data`` + # so that the following fields can read it + data.seek(-field.option_padding, io.SEEK_CUR) + consumed = length - field.option_padding + + self.__buffer__[field.name] = byte[:consumed] + packet['__length__'] -= consumed + else: + packet['__length__'] -= length + + if packet['__length__'] < 0: + warn(f'packet length < 0: {packet["__length__"]}', + SchemaWarning, stacklevel=stacklevel()) + + self.__updated__ = False + return self + + @classmethod + def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None': + """Prepare ``packet`` data for unpacking process. + + Args: + packet: packet data + + Note: + This method is expected to directly modify any data stored + in the ``packet`` and thus no return is required. + + """ + + def post_process(self, packet: 'dict[str, Any]') -> 'Schema': + """Revise ``schema`` data after packing and/or unpacking process. + + Args: + packet: Unpacked data. + + Returns: + Revised schema. + + """ + return self + + def __init_subclass__(cls, /, *args: 'Any', **kwargs: 'Any') -> 'None': + """Refuse to derive from a finalised schema. + + :func:`~pcapkit.utilities.compat.final` is a promise to the type + checker and nothing more: it records ``__final__`` on the class and + leaves the interpreter free to subclass it anyway. Every schema + :func:`schema_final` finalises carries a generated ``__init__`` built + from the :attr:`__fields__` that were declared at that moment, so a + subclass adding a field afterwards inherits a constructor that cannot + set it. This turns that promise into a rule the interpreter keeps. + + Args: + *args: Arbitrary positional arguments. + **kwargs: Arbitrary keyword arguments in class definition. + + Raises: + SchemaError: If any class in ``cls``'s ancestry carries the + ``__final__`` marker in its own ``__dict__``. + + """ + # NOTE: the whole ancestry rather than ``cls.__bases__``, and + # ``base.__dict__`` rather than ``getattr`` -- see + # :meth:`pcapkit.corekit.infoclass.Info.__init_subclass__`, which makes + # the same choices for the same reasons. + for base in cls.__mro__[1:]: + if base.__dict__.get('__final__'): + raise SchemaError(f'{cls.__name__}: cannot subclass {base.__name__}, ' + 'which is final') + + # NOTE: forwarded rather than swallowed, so that a class keyword nobody + # accepts still reaches ``object`` and fails there. + # :meth:`EnumSchema.__init_subclass__` is the one caller that reaches here + # with nothing: it consumes its own ``code`` keyword and discards the + # rest, which is why a stray class keyword is tolerated on an + # :class:`EnumSchema` subclass and rejected on a plain one. + super().__init_subclass__(*args, **kwargs) + + def __new__(cls, *args: '_VT', **kwargs: '_VT') -> 'Self': # pylint: disable=unused-argument + """Create a new instance. + + The class will try to automatically generate ``__init__`` method with + one keyword parameter per field in :attr:`__fields__`, which is + inspired by :pep:`557` (:mod:`dataclasses`). + + Note: + Whether ``cls`` still needs finalising is read from its *own* + ``__dict__``, never inherited (:issue:`1498`). Each + class is finalised on its first construction, whatever its + ancestors' state, so a subclass of a + :attr:`~pcapkit.corekit.infoclass.FinalisedState.BASE` ancestor + still gets its own ``__builtin__``, ``__excluded__`` and generated + ``__init__``, and a ``@final`` class under a ``BASE`` ancestor is + refused like any other. + + Args: + *args: Arbitrary positional arguments. + **kwargs: Arbitrary keyword arguments. + + Raises: + SchemaError: If ``cls`` was marked with + :func:`~pcapkit.utilities.compat.final` but never finalised by + :func:`schema_final`, i.e. it carries ``__final__`` in its own + ``__dict__`` without + :attr:`~pcapkit.corekit.infoclass.FinalisedState.FINAL`. + + """ + # NOTE: first instantiation is the earliest point a bare ``@final`` can + # be caught -- ``final`` runs after the class object exists, so + # :meth:`__init_subclass__` has already returned. Nested inside the NONE + # branch, which is already one-shot per class, so a finalised schema pays + # nothing for it. See :meth:`pcapkit.corekit.infoclass.Info.__new__`, + # which carries the full reasoning; on this side the missing generated + # ``__init__`` is the one built from :attr:`__fields__`. + # + # Both lookups read ``cls.__dict__`` rather than ``getattr``, because + # both markers are ordinary, inherited class attributes. The outer one: + # :func:`schema_final` writes ``BASE`` onto the class it finalises, and a + # subclass reading that state through inheritance was never finalised + # itself -- it would keep the ``__excluded__`` it copied at declaration, + # without the names :func:`schema_final` adds, and leak ``__map__``, + # ``__map_reverse__``, ``__buffer__`` and ``__updated__`` from + # :meth:`to_dict` (GitHub issue #1498). The inner one: an ancestor + # hand-marked bare ``@final`` passes its marker to every subclass + # declared before it landed, and ``getattr`` would blame such a subclass + # for a mismarking that is its ancestor's. + if cls.__dict__.get('__finalised__', FinalisedState.NONE) == FinalisedState.NONE: + if cls.__dict__.get('__final__'): + raise SchemaError(f'{cls.__name__}: marked final but never finalised, so it has no generated ' + '__init__; apply schema_final, which applies final itself, not final alone') + cls = schema_final(cls, _finalised=False) + self = super().__new__(cls) + + # NOTE: We define the ``__map__`` and ``__map_reverse__`` attributes + # here under ``self`` to avoid them being considered as class variables + # and thus being shared by all instances. + super().__setattr__(self, '__map__', {}) + super().__setattr__(self, '__map_reverse__', {}) + + # NOTE: We only create the attributes for the instance itself, + # to avoid creating shared attributes for the class. + super().__setattr__(self, '__buffer__', {name: b'' for name in self.__fields__.keys()}) + super().__setattr__(self, '__updated__', True) + + return self + + def __post_init__(self, packet: 'Optional[dict[str, Any]]' = None) -> 'None': + """Fill in the fields the caller left unset. + + Args: + packet: Packet data, as forwarded from the ``__packet__`` keyword + argument of the generated ``__init__``. The schema is packed + here only when one is given; see the note below. + + """ + for name, field in self.__fields__.items(): + # NOTE: Read with a fallback rather than by subscript, since the + # generated ``__init__`` is not the only caller: :meth:`from_dict` + # seeds only the keys its argument carries, so a field the caller left + # out is missing from ``__dict__`` entirely rather than holding + # ``NO_VALUE``. + # + # What is tested is ``NO_VALUE`` alone, not ``NO_VALUE`` or ``None``. + # This method fills in what the caller did not say, and a ``None`` the + # caller passed *is* something said: on an optional field it is the + # chosen value, meaning this packet does not carry the field. It is + # also what :meth:`unpack` stores for a + # :class:`~pcapkit.corekit.fields.misc.ConditionalField` whose test + # fails -- including one that declares a default of its own -- so + # substituting the default here would leave a constructed schema + # disagreeing with a parsed one about the same packet, and + # ``from_dict(parsed.to_dict())`` no longer reproducing what it was + # given. Telling the two apart is what ``NO_VALUE`` is for. + value = self.__dict__.get(name, NO_VALUE) + if value is not NO_VALUE: + continue + + default = field.default + if default is not NO_VALUE: + self.__dict__[name] = default + else: + # NOTE: Nothing to fill an unset field with, so the ``NO_VALUE`` + # the generated ``__init__`` seeded it with is dropped rather than + # kept: it is a *field* sentinel, not a value a schema may hold. + # Dropping it rather than storing ``None`` also keeps the name out + # of the context :meth:`pack` builds from ``__dict__``, which a + # schema may be relying on to seed for itself -- the PCAP-NG + # section header block reads its Byte-Order Magic from a ``match`` + # its own :meth:`pre_pack` supplies, and only when the context + # does not name one already. :meth:`pack` reads an absent field as + # ``None`` regardless. + self.__dict__.pop(name, None) + + # NOTE: Packed here only when a packet context was actually handed over. + # A schema is not in general packable from its own fields alone: a field + # callback may read a key that the *enclosing* layer owns, as the + # Multipath TCP options do with the ``length`` of the TCP option that + # carries them, and packing without it raises rather than producing + # octets. ``__updated__`` is still set, so a schema left unpacked here is + # packed by :meth:`__bytes__` on first use -- by which time the enclosing + # layer has supplied the context. + if packet is not None: + self.pack(packet) + + def __update__(self, dict_: 'Optional[Mapping[str, _VT] | Iterable[tuple[str, _VT]]]' = None, + **kwargs: '_VT') -> 'None': + # NOTE: Keys with the same names as the class's builtin methods will be + # renamed with the class name prefixed as mangled class variables + # implicitly and internally. Such mapping information will be stored + # within the ``__map__`` attribute. + + __name__ = type(self).__name__ # pylint: disable=redefined-builtin + + if dict_ is None: + data_iter = kwargs.items() # type: Iterable[tuple[str, Any]] + elif isinstance(dict_, collections.abc.Mapping): + data_iter = itertools.chain(dict_.items(), kwargs.items()) + else: + data_iter = itertools.chain(dict_, kwargs.items()) + + for (key, value) in data_iter: + # NOTE: Not a field, but what :meth:`to_dict` emits for the octets an + # enclosing ``SchemaField`` span held beyond this schema, and where, + # so that a rebuilt schema still re-emits them (#1380). + # NOTE: Nor is the short read :meth:`unpack` records (#1458). + if key in ('__remainder__', '__remainder_offset__', '__short_read__'): + self.__dict__[key] = value + continue + + if key not in self.__buffer__: + warn(f'{key!r} is not a valid field name', UnknownFieldWarning) + continue + + if key in self.__builtin__: + new_key = f'_{__name__}{key}' + + # NOTE: We keep record of the mapping bidirectionally. + self.__map__[key] = new_key + self.__map_reverse__[new_key] = key + + key = new_key + + # if key in self.__dict__: + # raise KeyExists(f'{key!r} already exists') + + # NOTE: We don't rewrite the key names here, just keep the + # original ones, even though they might break on the ``.`` + # (:obj:`getattr`) operator. + + # if isinstance(key, str): + # key = re.sub(r'\W', '_', key) + self.__dict__[key] = value + + self.__updated__ = True + + __init__ = __update__ + + def __str__(self) -> 'str': + temp = [] # type: list[str] + for (key, value) in self.__dict__.items(): + if key in self.__excluded__: + continue + + out_key = self.__map_reverse__.get(key, key) + temp.append(f'{out_key}={value}') + args = ', '.join(temp) + return f'{type(self).__name__}({args})' + + def __repr__(self) -> 'str': + temp = [] # type: list[str] + for (key, value) in self.__dict__.items(): + if key in self.__excluded__: + continue - # NOTE: bounded as the payload read above is (#1975). - byte = data.read(min(length, sys.maxsize)) - self.__buffer__[field.name] = byte - if not isinstance(field, ForwardMatchField): - self._note_short_read(field.name, length, byte) + out_key = self.__map_reverse__.get(key, key) + if isinstance(value, Schema): + temp.append(f'{out_key}={type(value).__name__}(...)') + else: + temp.append(f'{out_key}={value!r}') + args = ', '.join(temp) + return f'{type(self).__name__}({args})' - value = field.unpack(byte, packet.copy()) - if isinstance(field, SchemaField) and isinstance(value, Schema): - # NOTE: Kept on the nested schema rather than only in - # ``__buffer__``, so that the octets it left unread go wherever - # the value goes, :meth:`to_dict` included; see :meth:`pack`. - # The width read is summed from the nested buffers rather than - # taken with :func:`len`, which may pack the schema again. - consumed = sum(map(len, value.__buffer__.values())) - if len(byte) > consumed: - value.__dict__['__remainder__'] = byte[consumed:] - value.__dict__['__remainder_offset__'] = consumed - setattr(self, field.name, value) + def __bytes__(self) -> 'bytes': + if self.__updated__: + self.pack() - packet[field.name] = value + buffer = [] # type: list[bytes] + for name in self.__fields__.keys(): + value = self.__buffer__[name] + buffer.append(value) + return b''.join(buffer) - if isinstance(field, OptionField): - packet['__option_padding__'] = field.option_padding + def __len__(self) -> 'int': + return len(self.__bytes__()) - if isinstance(field, ForwardMatchField): - # NOTE: a forward match reads ``length`` octets so a later field - # can size itself from them, but consumes neither the stream - # (the rewind below) nor ``__length__`` (no decrement in this - # branch). ``self.__buffer__[field.name]`` above still holds the - # octets just read, so it is zeroed here, as :meth:`pack` does for - # the same field type: ``__bytes__``/``__len__`` concatenate every - # slot in ``__buffer__``, and the octets are read again, for real, - # by whichever field actually needs them, so keeping them would - # over-report the schema's length by the forward match's width. - # Zeroing is also what lets a declared area checked against - # ``len(self)`` -- :class:`~pcapkit.corekit.fields.collections.OptionField` - # and :class:`~pcapkit.corekit.fields.collections.ListField` both - # do this -- see the octets actually consumed. - # - # The rewind is by the octets actually read, not by ``length``: - # near the end of the stream the read comes back short, and - # rewinding the full width would move the stream *before* the - # forward match, so the fields after it would re-read the - # preceding octet as their first. - data.seek(-len(byte), io.SEEK_CUR) - self.__buffer__[field.name] = b'' - elif isinstance(field, OptionField) and field.option_padding > 0: - # the option list ended before the declared field length was - # exhausted; give the unconsumed remainder back to ``data`` - # so that the following fields can read it - data.seek(-field.option_padding, io.SEEK_CUR) - consumed = length - field.option_padding + def __iter__(self) -> 'Iterator[str]': + for key in self.__dict__: + if key in self.__builtin__: + continue + yield self.__map_reverse__.get(key, key) - self.__buffer__[field.name] = byte[:consumed] - packet['__length__'] -= consumed - else: - packet['__length__'] -= length + def __getitem__(self, name: 'str') -> '_VT': + # NOTE: Serves every key :meth:`__iter__` yields, not only the declared + # fields: a ``post_process`` may store derived values on the instance, + # as :class:`~pcapkit.protocols.schema.internet.ipv4.TSOption` does, and + # iteration reports them, so ``dict(schema)`` must be able to read them + # back (#1291). A mangled key is reachable only under the name + # it was mangled from, as in iteration. + if name in self.__fields__ or (name not in self.__builtin__ and name not in self.__map_reverse__): + key = self.__map__.get(name, name) + return self.__dict__[key] + raise KeyError(name) - if packet['__length__'] < 0: - warn(f'packet length < 0: {packet["__length__"]}', - SchemaWarning, stacklevel=stacklevel()) + def __setattr__(self, name: 'str', value: '_VT') -> 'None': + if name in self.__fields__: + key = self.__map__.get(name, name) + self.__dict__[key] = value + # NOTE: ``self.__updated__ = True`` would re-enter this method once + # per field assigned, only to miss the ``__fields__`` test and fall + # through to ``object.__setattr__``. ``__updated__`` is an instance + # attribute established in ``__new__``, so the direct store is the same + # write with none of the round trip. + self.__dict__['__updated__'] = True + return + return super().__setattr__(name, value) - self.__updated__ = False - return self + def __delattr__(self, name: 'str') -> 'None': + if name in self.__fields__: + key = self.__map__.get(name, name) + del self.__dict__[key] + self.__updated__ = True + return + return super().__delattr__(name) - @classmethod - def pre_unpack(cls, packet: 'dict[str, Any]') -> 'None': - """Prepare ``packet`` data for unpacking process. + def _note_short_read(self, name: 'str', length: 'int', byte: 'bytes') -> 'None': + """Record the first field the data ends inside, for :meth:`_cut_short_read`. Args: - packet: packet data - - Note: - This method is expected to directly modify any data stored - in the ``packet`` and thus no return is required. + name: Field name. + length: Octets the field declares. + byte: Octets actually read for it. """ + if len(byte) < length and '__short_read__' not in self.__dict__: + self.__dict__['__short_read__'] = (name, len(byte)) - def post_process(self, packet: 'dict[str, Any]') -> 'Schema': - """Revise ``schema`` data after packing and/or unpacking process. - - Args: - packet: Unpacked data. + def _cut_short_read(self) -> 'None': + """Cut the packed buffers back to the octets a short read consumed. - Returns: - Revised schema. + A schema whose data ended inside one of its fields records, as + ``__short_read__``, that field's name and how many of its octets were + there (see :meth:`unpack`). Its value is read as though the missing + octets were zeros, so packing it writes the field at full width and + every field after it besides, none of which was captured. Only the + octets read are kept here, so a truncated header packs back to exactly + what was captured (:issue:`1458`). """ - return self + record = self.__dict__.get('__short_read__') + if record is None: + return + + name, count = record + cut = False + for key in self.__fields__: + if cut: + self.__buffer__[key] = b'' + elif key == name: + self.__buffer__[key] = self.__buffer__[key][:count] + cut = True class _EnumRegistry(collections.defaultdict): @@ -1361,6 +1361,53 @@ def registry(self) -> 'DefaultDict[_ET, Type[Self]]': """ return self.__enum__ + @classmethod + def register(cls, code: '_ET', schema: 'Type[Self]') -> 'None': + """Register enumeration to :attr:`__enum__` mapping. + + Args: + code: Enumeration code. + schema: Enumeration schema. + + Warns: + pcapkit.utilities.warnings.RegistryWarning: If ``code`` is already + registered, naming the displaced schema and its replacement. + + Note: + Every public registrar in + :mod:`pcapkit.foundation.registry.protocols` that accepts a + ``schema`` registers two halves of one binding -- a parser class + through e.g. :meth:`IPv4.register_option + `, and a + schema class through this method. Both warn on an overwrite, so one + ``register_ipv4_option`` call replacing a built-in reports the + schema it displaced as well as the parser. + + The warning fires only when the incumbent differs from the + replacement: a repeat call that names the exact same schema object + is a caller replaying a registration, not a mistake, so it is a + silent no-op. + + Presence is a faithful "was this really registered" test only because + :class:`_EnumRegistry` returns a miss without recording it. A plain + :class:`collections.defaultdict` would insert :attr:`__default__` the + first time any unregistered ``code`` was looked up, so parsing a + single packet carrying an unknown code would make the next legitimate + registration for that code warn about an entry no caller ever asked + for. + + :class:`pcapkit.protocols.schema.misc.pcapng.Option` overrides this + method with a namespaced registry of its own and does not delegate + here, so it is guarded separately. + + """ + incumbent = cls.__enum__.get(code) # type: ignore[arg-type] + if incumbent is not None and incumbent is not schema: + warn(f'schema {code} already registered, overwriting ' + f'{incumbent!r} with {schema!r}', RegistryWarning) + + cls.__enum__[code] = schema # type: ignore[index] + def __init_subclass__(cls, /, code: 'Optional[_ET | Iterable[_ET]]' = None, *args: 'Any', **kwargs: 'Any') -> 'None': """Register enumeration to :attr:`registry` mapping. @@ -1441,53 +1488,6 @@ def __init_subclass__(cls, /, code: 'Optional[_ET | Iterable[_ET]]' = None, *arg f'{incumbent!r} with {cls!r}', RegistryWarning) cls.__enum__[_code] = (cls) # type: ignore[index] - @classmethod - def register(cls, code: '_ET', schema: 'Type[Self]') -> 'None': - """Register enumeration to :attr:`__enum__` mapping. - - Args: - code: Enumeration code. - schema: Enumeration schema. - - Warns: - pcapkit.utilities.warnings.RegistryWarning: If ``code`` is already - registered, naming the displaced schema and its replacement. - - Note: - Every public registrar in - :mod:`pcapkit.foundation.registry.protocols` that accepts a - ``schema`` registers two halves of one binding -- a parser class - through e.g. :meth:`IPv4.register_option - `, and a - schema class through this method. Both warn on an overwrite, so one - ``register_ipv4_option`` call replacing a built-in reports the - schema it displaced as well as the parser. - - The warning fires only when the incumbent differs from the - replacement: a repeat call that names the exact same schema object - is a caller replaying a registration, not a mistake, so it is a - silent no-op. - - Presence is a faithful "was this really registered" test only because - :class:`_EnumRegistry` returns a miss without recording it. A plain - :class:`collections.defaultdict` would insert :attr:`__default__` the - first time any unregistered ``code`` was looked up, so parsing a - single packet carrying an unknown code would make the next legitimate - registration for that code warn about an entry no caller ever asked - for. - - :class:`pcapkit.protocols.schema.misc.pcapng.Option` overrides this - method with a namespaced registry of its own and does not delegate - here, so it is guarded separately. - - """ - incumbent = cls.__enum__.get(code) # type: ignore[arg-type] - if incumbent is not None and incumbent is not schema: - warn(f'schema {code} already registered, overwriting ' - f'{incumbent!r} with {schema!r}', RegistryWarning) - - cls.__enum__[code] = schema # type: ignore[index] - def keep_short_read(protocol: 'ProtocolBase') -> 'None': """Carry a parsed header's short read into the protocol's info.