Skip to content

Repair the PCAP-NG write path (#365-#368) #388

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JarryShaw merged 1 commit into from
Sep 15, 2026
Merged

Repair the PCAP-NG write path (#365-#368) #388

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions pcapkit/corekit/fields/numbers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,8 +142,19 @@ def pre_process(self, value: 'int', packet: 'dict[str, Any]') -> 'int | bytes':
Returns:
Processed field value.

Notes:
Masking against :attr:`self._bit_mask <NumberField.bit_length>`
truncates the value to the field's bit length, but it also turns a
negative value into its unsigned two's-complement pattern, which
neither :func:`struct.pack` nor :meth:`int.to_bytes` accepts for a
signed field. A signed field therefore maps the pattern back into
its signed range afterwards, so that e.g. a PCAP-NG section length
of ``-1`` (section length not specified) can be written out.

"""
value = value & self._bit_mask
if self._signed and value > self._bit_mask >> 1:
value -= self._bit_mask + 1
if not self._need_process:
return value

Expand Down
20 changes: 9 additions & 11 deletions pcapkit/foundation/engines/pcapng.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,13 +27,11 @@
from pcapkit.protocols.data.misc.pcapng import CustomBlock as Data_CustomBlock
from pcapkit.protocols.data.misc.pcapng import \
DecryptionSecretsBlock as Data_DecryptionSecretsBlock
from pcapkit.protocols.data.misc.pcapng import EnhancedPacketBlock as Data_EnhancedPacketBlock
from pcapkit.protocols.data.misc.pcapng import \
InterfaceDescriptionBlock as Data_InterfaceDescriptionBlock
from pcapkit.protocols.data.misc.pcapng import \
InterfaceStatisticsBlock as Data_InterfaceStatisticsBlock
from pcapkit.protocols.data.misc.pcapng import NameResolutionBlock as Data_NameResolutionBlock
from pcapkit.protocols.data.misc.pcapng import PacketBlock as Data_PacketBlock
from pcapkit.protocols.data.misc.pcapng import SectionHeaderBlock as Data_SectionHeaderBlock
from pcapkit.protocols.data.misc.pcapng import \
SystemdJournalExportBlock as Data_SystemdJournalExportBlock
Expand Down Expand Up @@ -214,9 +212,12 @@ def read_frame(self) -> 'P_PCAPNG':
self._write_file(block.info, name=f'Decryption Secrets {len(self._ctx.secrets)}')

elif block.info.type == Enum_BlockType.Interface_Statistics_Block:
# NOTE: The interface ID is bounds-checked while the block is
# parsed, by ``PCAPNG._get_interface``. Re-checking it here
# cannot fire: parsing an ISB resolves its interface to read the
# block's timestamp, so a block that reaches this branch has
# already been validated.
isb_info = cast('Data_InterfaceStatisticsBlock', block.info)
if isb_info.interface_id >= len(self._ctx.interfaces):
raise FormatError(f'PCAP-NG: [ISB] invalid interface ID: {isb_info.interface_id}')
self._ctx.statistics.append(isb_info)

self._write_file(isb_info, name=f'Interface Statistics {len(self._ctx.statistics)}')
Expand All @@ -227,9 +228,8 @@ def read_frame(self) -> 'P_PCAPNG':
self._write_file(block.info, name=f'Custom {len(self._ctx.custom)}')

elif block.info.type == Enum_BlockType.Enhanced_Packet_Block:
epb_info = cast('Data_EnhancedPacketBlock', block.info)
if epb_info.interface_id >= len(self._ctx.interfaces):
raise FormatError(f'PCAP-NG: [EPB] invalid interface ID: {epb_info.interface_id}')
# NOTE: as for the ISB above, the interface ID has already been
# bounds-checked by ``PCAPNG._get_interface`` while parsing.
break

elif block.info.type == Enum_BlockType.Simple_Packet_Block:
Expand All @@ -242,10 +242,8 @@ def read_frame(self) -> 'P_PCAPNG':
break

elif block.info.type == Enum_BlockType.Packet_Block:
pack_info = cast('Data_PacketBlock', block.info)
if pack_info.interface_id >= len(self._ctx.interfaces):
raise FormatError(f'PCAP-NG: [Packet] invalid interface ID: {pack_info.interface_id}')

# NOTE: as for the ISB above, the interface ID has already been
# bounds-checked by ``PCAPNG._get_interface`` while parsing.
warn('PCAP-NG: [Packet] deprecated block type', DeprecatedFormatWarning,
stacklevel=stacklevel())
break
Expand Down
75 changes: 60 additions & 15 deletions pcapkit/protocols/misc/pcapng.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,7 +175,8 @@
from pcapkit.protocols.schema.misc.pcapng import ZigBeeNWKKey as Schema_ZigBeeNWKKey
from pcapkit.protocols.schema.schema import Schema
from pcapkit.utilities.compat import StrEnum, localcontext
from pcapkit.utilities.exceptions import ProtocolError, RegistryError, UnsupportedCall, stacklevel
from pcapkit.utilities.exceptions import (FormatError, ProtocolError, RegistryError,
UnsupportedCall, stacklevel)
from pcapkit.utilities.warnings import (AttributeWarning, DeprecatedFormatWarning, ProtocolWarning,
RegistryWarning, warn)

Expand Down Expand Up @@ -532,6 +533,17 @@ class PCAPNG(Protocol[Data_PCAPNG, Schema_PCAPNG],
Enum_BlockType.Simple_Packet_Block,
Enum_BlockType.Packet_Block)

#: Blocks that resolve an interface of their section, mapped to the tag used
#: when reporting an interface ID that names no such interface. A Simple
#: Packet Block carries no interface ID field and always refers to the
#: section's first interface, so it is listed here too.
INTERFACE_ID_BLOCK_TAGS = {
Enum_BlockType.Enhanced_Packet_Block: 'EPB',
Enum_BlockType.Simple_Packet_Block: 'SPB',
Enum_BlockType.Packet_Block: 'Packet',
Enum_BlockType.Interface_Statistics_Block: 'ISB',
} # type: dict[int, str]

##########################################################################
# Defaults.
##########################################################################
Expand Down Expand Up @@ -1119,6 +1131,33 @@ def _get_local_timezone() -> 'timezone':
return datetime.timezone.utc
return cast('timezone', tzinfo)

def _get_interface(self, interface_id: 'int') -> 'Data_InterfaceDescriptionBlock':
"""Interface description that ``interface_id`` names.

Args:
interface_id: Interface ID that the current block associates with.

Returns:
Interface Description Block (IDB) of the current section that
``interface_id`` identifies.

Raises:
FormatError: If the current section describes no such interface.

Note:
The PCAP-NG specification requires a block's interface ID to name an
Interface Description Block of the block's own section, so an ID that
names none is a malformed file rather than a programming error --
hence :exc:`~pcapkit.utilities.exceptions.FormatError` rather than
the bare :exc:`IndexError` that indexing the list would raise.

"""
interfaces = self._ctx.interfaces
if not 0 <= interface_id < len(interfaces):
tag = self.INTERFACE_ID_BLOCK_TAGS.get(self._type, f'Block {self._type}')
raise FormatError(f'PCAP-NG: [{tag}] invalid interface ID: {interface_id}')
return interfaces[interface_id]

def _get_resolution(self, interface_id: 'int' = 0) -> 'int':
"""Timestamp resolution of the current block, in units per second.

Expand All @@ -1135,7 +1174,7 @@ def _get_resolution(self, interface_id: 'int' = 0) -> 'int':
AttributeWarning, stacklevel=stacklevel())
return 1_000_000

options = self._ctx.interfaces[interface_id].options
options = self._get_interface(interface_id).options
tsresol = cast('Optional[Data_IF_TSResolOption]',
options.get(Enum_OptionType.if_tsresol))
if tsresol is None:
Expand All @@ -1158,7 +1197,7 @@ def _get_offset(self, interface_id: 'int' = 0) -> 'int':
AttributeWarning, stacklevel=stacklevel())
return 0

options = self._ctx.interfaces[interface_id].options
options = self._get_interface(interface_id).options
tsoffset = cast('Optional[Data_IF_TSOffsetOption]',
options.get(Enum_OptionType.if_tsoffset))
if tsoffset is None:
Expand All @@ -1181,7 +1220,7 @@ def _get_timezone(self, interface_id: 'int' = 0) -> 'timezone':
AttributeWarning, stacklevel=stacklevel())
return self._get_local_timezone()

options = self._ctx.interfaces[interface_id].options
options = self._get_interface(interface_id).options
tzone = cast('Optional[Data_IF_TZoneOption]',
options.get(Enum_OptionType.if_tzone))
if tzone is None:
Expand All @@ -1204,7 +1243,7 @@ def _get_linktype(self, interface_id: 'int' = 0) -> 'Enum_LinkType':
"""
if self._ctx is None or self._type not in self.PACKET_TYPES:
raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute '_get_linktype'")
return self._ctx.interfaces[interface_id].linktype
return self._get_interface(interface_id).linktype

def _read_timestamp(self, timestamp_high: 'int', timestamp_low: 'int', *,
interface_id: 'int' = 0) -> 'tuple[dt_type, Decimal]':
Expand Down Expand Up @@ -3338,7 +3377,7 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *,
minor_version = version[1]

if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='shb')
options_value, total_length = self._make_pcapng_options(options, namespace='opt')
else:
options_value, total_length = [], 0

Expand Down Expand Up @@ -3385,7 +3424,7 @@ def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = No
reversed=linktype_reversed, pack=False)

if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='shb')
options_value, total_length = self._make_pcapng_options(options, namespace='if')
else:
options_value, total_length = [], 0

Expand Down Expand Up @@ -3433,14 +3472,14 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *,
if self._ctx is None:
snaplen = 0xFFFF_FFFF_FFFF_FFFF
else:
snaplen = self._ctx.interfaces[interface_id].snaplen
snaplen = self._get_interface(interface_id).snaplen
captured_len = min(len(packet_data), snaplen)
if original_len is None:
original_len = len(packet_data)
packet_len = math.ceil(len(packet_data) / 4) * 4

if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='shb')
options_value, total_length = self._make_pcapng_options(options, namespace='epb')
else:
options_value, total_length = [], 0

Expand Down Expand Up @@ -3512,7 +3551,7 @@ def _make_block_nrb(self, block: 'Optional[Data_NameResolutionBlock]' = None, *,
records_value, records_length = [], 0

if options is not None:
options_value, options_length = self._make_pcapng_options(options, namespace='shb')
options_value, options_length = self._make_pcapng_options(options, namespace='ns')
else:
options_value, options_length = [], 0

Expand Down Expand Up @@ -3549,7 +3588,7 @@ def _make_block_isb(self, block: 'Optional[Data_InterfaceStatisticsBlock]' = Non
timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id)

if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='shb')
options_value, total_length = self._make_pcapng_options(options, namespace='isb')
else:
options_value, total_length = [], 0

Expand Down Expand Up @@ -3665,7 +3704,7 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None,
secrets_length = len(secrets_data_val)

if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='shb')
options_value, total_length = self._make_pcapng_options(options, namespace='dsb')
else:
options_value, total_length = [], 0

Expand Down Expand Up @@ -3701,8 +3740,14 @@ def _make_block_cb(self, block: 'Optional[Data_CustomBlock]' = None, *,
data = block.data
options = cast('Option', getattr(block, 'options', None))

# NOTE: Unlike every other block, a custom block has no ``OptionField``
# to name an option registry: :class:`~pcapkit.protocols.schema.misc.pcapng.CustomBlock`
# spans the custom data, its padding and the options as one opaque blob,
# since only the owner of the private enterprise number knows where the
# custom data ends. Its options are therefore resolved in the generic
# ``opt_*`` space, which is the only one valid in every block.
if options is not None:
options_value, _ = self._make_pcapng_options(options, namespace='shb')
options_value, _ = self._make_pcapng_options(options, namespace='opt')
else:
options_value, _ = [], 0

Expand Down Expand Up @@ -3765,14 +3810,14 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *,
if self._ctx is None:
snaplen = 0xFFFF_FFFF_FFFF_FFFF
else:
snaplen = self._ctx.interfaces[interface_id].snaplen
snaplen = self._get_interface(interface_id).snaplen
captured_len = min(len(packet_data), snaplen)
if original_len is None:
original_len = len(packet_data)
packet_len = math.ceil(len(packet_data) / 4) * 4

if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='shb')
options_value, total_length = self._make_pcapng_options(options, namespace='pack')
else:
options_value, total_length = [], 0

Expand Down
Loading