Skip to content

fix(pcapng): blocks and options rebuild as parsed (#1272, #1273, #1277, #1278, #1281, #1286) #1378

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 2 commits into from
Oct 8, 2026
Merged

fix(pcapng): blocks and options rebuild as parsed (#1272, #1273, #1277, #1278, #1281, #1286) #1378

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
497a3e9
fix(pcapng): blocks and options rebuild as parsed (#1272, #1273, #127…
JarryShaw Oct 7, 2026
d05e444
Merge branch 'main' into fix/pcapng-roundtrip-modelling
JarryShaw Oct 8, 2026
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
12 changes: 10 additions & 2 deletions pcapkit/protocols/data/misc/pcapng.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -353,6 +353,10 @@ class EPB_FlagsOption(Option):
reception: 'PacketReception'
#: FCS length.
fcs_len: 'int'
#: Bits 9 to 23 of the flags word, kept verbatim (checksum and
#: segmentation offload flags, reserved bits and unnamed
#: link-layer-dependent errors).
reserved: 'int'
#: Link-layer-dependent error - CRC error (bit 24).
crc_error: 'bool'
#: Link-layer-dependent error - packet too long error (bit 25).
Expand All @@ -372,7 +376,7 @@ class EPB_FlagsOption(Option):

if TYPE_CHECKING:
def __init__(self, type: 'Enum_OptionType', length: 'int', direction: 'PacketDirection',
reception: 'PacketReception', fcs_len: 'int', crc_error: 'bool',
reception: 'PacketReception', fcs_len: 'int', reserved: 'int', crc_error: 'bool',
too_long: 'bool', too_short: 'bool', gap_error: 'bool', unaligned_error: 'bool',
delimiter_error: 'bool', preamble_error: 'bool', symbol_error: 'bool') -> 'None': ...

Expand Down Expand Up @@ -861,6 +865,10 @@ class PACK_FlagsOption(Option):
reception: 'PacketReception'
#: FCS length.
fcs_len: 'int'
#: Bits 9 to 23 of the flags word, kept verbatim (checksum and
#: segmentation offload flags, reserved bits and unnamed
#: link-layer-dependent errors).
reserved: 'int'
#: Link-layer-dependent error - CRC error (bit 24).
crc_error: 'bool'
#: Link-layer-dependent error - packet too long error (bit 25).
Expand All @@ -880,7 +888,7 @@ class PACK_FlagsOption(Option):

if TYPE_CHECKING:
def __init__(self, type: 'Enum_OptionType', length: 'int', direction: 'PacketDirection',
reception: 'PacketReception', fcs_len: 'int', crc_error: 'bool',
reception: 'PacketReception', fcs_len: 'int', reserved: 'int', crc_error: 'bool',
too_long: 'bool', too_short: 'bool', gap_error: 'bool', unaligned_error: 'bool',
delimiter_error: 'bool', preamble_error: 'bool', symbol_error: 'bool') -> 'None': ...

Expand Down
90 changes: 77 additions & 13 deletions pcapkit/protocols/misc/pcapng.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,6 +262,28 @@ def _utf8_length(text: 'str') -> 'int':
return len(text.encode('utf-8'))


def _join_names(names: 'list[str]') -> 'str':
"""Join NRB record names into zero-terminated name resolution data.

Args:
names: Host names; a :class:`~pcapkit.corekit.fields.strings.DecodedString`
contributes the octets it was decoded from.

Returns:
The name resolution data, as a
:class:`~pcapkit.corekit.fields.strings.DecodedString` when any name
carries octets that its text does not encode back to.

"""
names = list(names) or ['']
text = ''.join(f'{name}\x00' for name in names)
raw = b''.join((name.raw if isinstance(name, DecodedString) else name.encode('utf-8')) + b'\x00'
for name in names)
if text.encode('utf-8') == raw:
return text
return DecodedString(text, raw)


class PacketDirection(EnumLookup, enum.IntEnum):
"""Packet direction for ``epb_flags`` options.

Expand Down Expand Up @@ -1020,6 +1042,10 @@ def pack(self, **kwargs: 'Any') -> 'bytes':
if self._ctx is not None:
self._byte = self._ctx.section.byteorder
packet['byteorder'] = self._byte
elif self._type == Enum_BlockType.Section_Header_Block:
# NOTE: A section header starts its own section, so the byte order
# is the one :meth:`_make_block_shb` chose rather than the host's.
packet.setdefault('byteorder', self._byte)
return self.__header__.pack(packet)

def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_PCAPNG':
Expand Down Expand Up @@ -1600,6 +1626,28 @@ def _make_data(cls, data: 'Data_PCAPNG') -> 'dict[str, Any]': # type: ignore[ov
'block': data,
}

@staticmethod
def _flags_enum(enum: 'Type[PacketDirection] | Type[PacketReception]', value: 'int',
option: 'Enum_OptionType') -> 'Any':
"""Look a flags word sub-field up in its enumeration.

Args:
enum: :class:`PacketDirection` or :class:`PacketReception`.
value: Sub-field value.
option: Option type, for the error message.

Returns:
The enumeration member.

Raises:
ProtocolError: If the specification assigns ``value`` no meaning.

"""
try:
return enum(value)
except ValueError:
raise ProtocolError(f'PCAP-NG: [{option.name}] invalid {enum.__name__}: {value:#b}') from None

@classmethod
def _make_packet_data(cls, block: 'Data_PCAPNG') -> 'bytes | ProtocolBase':
"""Create the packet data of an EPB, SPB or PB from its data model.
Expand Down Expand Up @@ -2920,9 +2968,10 @@ def _read_option_epb_flags(self, schema: 'Schema_EPB_FlagsOption', *,
option = Data_EPB_FlagsOption(
type=schema.type,
length=schema.length,
direction=PacketDirection(schema.flags['direction']),
reception=PacketReception(schema.flags['reception']),
direction=self._flags_enum(PacketDirection, schema.flags['direction'], schema.type),
reception=self._flags_enum(PacketReception, schema.flags['reception'], schema.type),
fcs_len=schema.flags['fcs_len'],
reserved=schema.flags['reserved'],
crc_error=bool(schema.flags['crc_error']),
too_long=bool(schema.flags['too_long']),
too_short=bool(schema.flags['too_short']),
Expand Down Expand Up @@ -3378,9 +3427,10 @@ def _read_option_pack_flags(self, schema: 'Schema_PACK_FlagsOption', *,
option = Data_PACK_FlagsOption(
type=schema.type,
length=schema.length,
direction=PacketDirection(schema.flags['direction']),
reception=PacketReception(schema.flags['reception']),
direction=self._flags_enum(PacketDirection, schema.flags['direction'], schema.type),
reception=self._flags_enum(PacketReception, schema.flags['reception'], schema.type),
fcs_len=schema.flags['fcs_len'],
reserved=schema.flags['reserved'],
crc_error=bool(schema.flags['crc_error']),
too_long=bool(schema.flags['too_long']),
too_short=bool(schema.flags['too_short']),
Expand Down Expand Up @@ -3696,7 +3746,10 @@ def _make_block_unknown(self, block: 'Optional[Data_UnknownBlock]' = None, *,
if block is not None:
data = block.body

length = len(data)
# NOTE: The body is padded to 32 bits, and Block Total Length counts it
# together with the 12 octets of block type and both length fields.
data += bytes(-len(data) % 4)
length = len(data) + 12

return Schema_UnknownBlock(
length=length,
Expand All @@ -3710,6 +3763,7 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *,
minor_version: 'Optional[int]' = None,
section_length: 'int' = -1,
options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None,
byteorder: 'Literal["big", "little"]' = sys.byteorder,
**kwargs: 'Any') -> 'Schema_SectionHeaderBlock':
"""Make PCAP-NG section header block (SHB).

Expand All @@ -3720,13 +3774,16 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *,
minor_version: Minor version number.
section_length: Section length.
options: Block options.
byteorder: Byte order of the section; a parsed block keeps its own.
**kwargs: Arbitrary keyword arguments.

Returns:
Constructed block schema.

"""
self._byte = sys.byteorder
if block is not None:
byteorder = block.byteorder
self._byte = byteorder

if block is not None:
major_version = block.version.major
Expand Down Expand Up @@ -4958,6 +5015,7 @@ def _make_option_epb_flags(self, type: 'Enum_OptionType', option: 'Optional[Data
reception_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
reception_reversed: 'bool' = False,
fcs_len: 'int' = 0,
reserved: 'int' = 0,
crc_error: 'bool' = False,
too_long: 'bool' = False,
too_short: 'bool' = False,
Expand All @@ -4981,6 +5039,7 @@ def _make_option_epb_flags(self, type: 'Enum_OptionType', option: 'Optional[Data
reception_namespace: Namespace of packet reception.
reception_reversed: Whether to reverse packet reception namespace.
fcs_len: Length of FCS field, in bytes.
reserved: Bits 9 to 23 of the flags word, verbatim.
crc_error: Whether CRC error occurred.
too_long: Whether packet is too long.
too_short: Whether packet is too short.
Expand All @@ -5006,6 +5065,7 @@ def _make_option_epb_flags(self, type: 'Enum_OptionType', option: 'Optional[Data
direction_val = option.direction
reception_val = option.reception
fcs_len = option.fcs_len
reserved = option.reserved
crc_error = option.crc_error
too_long = option.too_long
too_short = option.too_short
Expand All @@ -5027,6 +5087,7 @@ def _make_option_epb_flags(self, type: 'Enum_OptionType', option: 'Optional[Data
'direction': getattr(direction_val, 'value', direction_val),
'reception': getattr(reception_val, 'value', reception_val),
'fcs_len': fcs_len,
'reserved': reserved,
'crc_error': int(crc_error),
'too_long': int(too_long),
'too_short': int(too_short),
Expand Down Expand Up @@ -5198,9 +5259,8 @@ def _make_option_epb_verdict(self, type: 'Enum_OptionType', option: 'Optional[Da
if self._type != Enum_BlockType.Enhanced_Packet_Block:
raise ProtocolError(f'PCAP-NG: [epb_verdict] option must be in Enhanced Packet Block, '
f'but found in {self._type} block.')
if self._opt[type] > 0:
raise ProtocolError(f'PCAP-NG: [epb_verdict] option must be only one, '
f'but {self._opt[type] + 1} found.')
# NOTE: No only-one guard: the specification allows ``epb_verdict`` to
# appear multiple times, and the parser accepts every occurrence.

if option is not None:
verdict_val = option.verdict
Expand Down Expand Up @@ -5540,6 +5600,7 @@ def _make_option_pack_flags(self, type: 'Enum_OptionType', option: 'Optional[Dat
reception_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long
reception_reversed: 'bool' = False,
fcs_len: 'int' = 0,
reserved: 'int' = 0,
crc_error: 'bool' = False,
too_long: 'bool' = False,
too_short: 'bool' = False,
Expand All @@ -5563,6 +5624,7 @@ def _make_option_pack_flags(self, type: 'Enum_OptionType', option: 'Optional[Dat
reception_namespace: Namespace of packet reception.
reception_reversed: Whether to reverse packet reception namespace.
fcs_len: Length of FCS field, in bytes.
reserved: Bits 9 to 23 of the flags word, verbatim.
crc_error: Whether CRC error occurred.
too_long: Whether packet is too long.
too_short: Whether packet is too short.
Expand All @@ -5588,6 +5650,7 @@ def _make_option_pack_flags(self, type: 'Enum_OptionType', option: 'Optional[Dat
direction_val = option.direction
reception_val = option.reception
fcs_len = option.fcs_len
reserved = option.reserved
crc_error = option.crc_error
too_long = option.too_long
too_short = option.too_short
Expand All @@ -5609,6 +5672,7 @@ def _make_option_pack_flags(self, type: 'Enum_OptionType', option: 'Optional[Dat
'direction': getattr(direction_val, 'value', direction_val),
'reception': getattr(reception_val, 'value', reception_val),
'fcs_len': fcs_len,
'reserved': reserved,
'crc_error': int(crc_error),
'too_long': int(too_long),
'too_short': int(too_short),
Expand Down Expand Up @@ -5809,11 +5873,11 @@ def _make_record_ipv4(self, type: 'Enum_RecordType', record: 'Optional[Data_IPv4

if names is None:
names = []
records = '\x00'.join(names) + '\x00'
records = _join_names(names)

return Schema_IPv4Record(
type=type,
length=4 + len(records),
length=4 + _utf8_length(records),
ip=ip,
resol=records,
)
Expand Down Expand Up @@ -5841,11 +5905,11 @@ def _make_record_ipv6(self, type: 'Enum_RecordType', record: 'Optional[Data_IPv6

if names is None:
names = []
records = '\x00'.join(names) + '\x00'
records = _join_names(names)

return Schema_IPv6Record(
type=type,
length=16 + len(records),
length=16 + _utf8_length(records),
ip=ip,
resol=records,
)
Expand Down
Loading
Loading