Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
66 changes: 41 additions & 25 deletions pcapkit/protocols/misc/pcapng/protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -3917,15 +3917,17 @@ def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = No
options_value, total_length = self._make_pcapng_options(options, namespace='if')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
total_length += self._make_options_trailer(block)

return self._make_padding(Schema_InterfaceDescriptionBlock(
schema = self._make_padding(Schema_InterfaceDescriptionBlock(
length=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Description_Block],
linktype=linktype_val,
snaplen=snaplen,
options=options_value,
length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Description_Block],
), block, 'reserved')
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(schema, block, 'padding', 'padding_opts')

def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *,
interface_id: 'int' = 0,
Expand Down Expand Up @@ -4001,9 +4003,9 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *,
options_value, total_length = self._make_pcapng_options(options, namespace='pack')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
total_length += self._make_options_trailer(block)

return self._make_padding(Schema_PacketBlock(
schema = self._make_padding(Schema_PacketBlock(
length=total_length + BLOCK_MINIMUM[Enum_BlockType.Packet_Block] + packet_len,
interface_id=interface_id,
drop_count=drop_count,
Expand All @@ -4015,6 +4017,8 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *,
options=options_value,
length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Packet_Block] + packet_len,
), block, 'padding_data')
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(schema, block, 'padding_opts')

def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *,
original_len: 'Optional[int]' = None,
Expand Down Expand Up @@ -4089,14 +4093,15 @@ def _make_block_nrb(self, block: 'Optional[Data_NameResolutionBlock]' = None, *,
options_value, options_length = self._make_pcapng_options(options, namespace='ns')
else:
options_value, options_length = [], 0
options_length += self._make_options_trailer(block, options_value)
options_length += self._make_options_trailer(block)

return Schema_NameResolutionBlock(
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(Schema_NameResolutionBlock(
length=options_length + records_length + GENERIC_BLOCK_MINIMUM,
records=records_value,
options=options_value,
length2=options_length + records_length + GENERIC_BLOCK_MINIMUM,
)
), block, 'padding', 'padding_opts')

def _make_block_isb(self, block: 'Optional[Data_InterfaceStatisticsBlock]' = None, *,
interface_id: 'int' = 0,
Expand Down Expand Up @@ -4130,16 +4135,17 @@ def _make_block_isb(self, block: 'Optional[Data_InterfaceStatisticsBlock]' = Non
options_value, total_length = self._make_pcapng_options(options, namespace='isb')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
total_length += self._make_options_trailer(block)

return Schema_InterfaceStatisticsBlock(
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(Schema_InterfaceStatisticsBlock(
length=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Statistics_Block],
interface_id=interface_id,
timestamp_high=timestamp_high,
timestamp_low=timestamp_low,
options=options_value,
length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Interface_Statistics_Block],
)
), block, 'padding', 'padding_opts')

def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *,
interface_id: 'int' = 0,
Expand Down Expand Up @@ -4196,9 +4202,9 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *,
options_value, total_length = self._make_pcapng_options(options, namespace='epb')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
total_length += self._make_options_trailer(block)

return self._make_padding(Schema_EnhancedPacketBlock(
schema = self._make_padding(Schema_EnhancedPacketBlock(
length=total_length + BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block] + packet_len,
interface_id=interface_id,
timestamp_high=timestamp_high,
Expand All @@ -4209,6 +4215,8 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *,
options=options_value,
length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Enhanced_Packet_Block] + packet_len,
), block, 'padding_data')
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(schema, block, 'padding_opts')

def _make_block_systemd(self, block: 'Optional[Data_SystemdJournalExportBlock]' = None, *,
entries: 'Optional[list[OrderedMultiDict[str, str | bytes]] | bytes]' = None,
Expand Down Expand Up @@ -4344,16 +4352,18 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None,
options_value, total_length = self._make_pcapng_options(options, namespace='dsb')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
total_length += self._make_options_trailer(block)

return self._make_padding(Schema_DecryptionSecretsBlock(
schema = self._make_padding(Schema_DecryptionSecretsBlock(
length=total_length + secrets_padded + BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block],
secrets_type=secrets_type_val,
secrets_length=secrets_length,
secrets_data=secrets_data_val,
options=options_value,
length2=total_length + secrets_padded + BLOCK_MINIMUM[Enum_BlockType.Decryption_Secrets_Block],
), block, 'padding_data')
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(schema, block, 'padding_opts')

def _make_block_cb(self, block: 'Optional[Data_CustomBlock]' = None, *,
pen: 'int' = 0,
Expand Down Expand Up @@ -4472,17 +4482,18 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *,
options_value, total_length = self._make_pcapng_options(options, namespace='opt')
else:
options_value, total_length = [], 0
total_length += self._make_options_trailer(block, options_value)
total_length += self._make_options_trailer(block)

return Schema_SectionHeaderBlock(
# NOTE: The octets after ``opt_endofopt``, if any (#1437, #2776).
return self._make_padding(Schema_SectionHeaderBlock(
length=total_length + BLOCK_MINIMUM[Enum_BlockType.Section_Header_Block],
magic=BYTE_ORDER_MAGIC,
major=major_version,
minor=minor_version,
section_length=section_length,
options=options_value,
length2=total_length + BLOCK_MINIMUM[Enum_BlockType.Section_Header_Block],
)
), block, 'padding', 'padding_opts')

def _make_option_unknown(self, type: 'Enum_OptionType', option: 'Optional[Data_UnknownOption]' = None, *,
data: 'bytes' = b'',
Expand Down Expand Up @@ -6548,26 +6559,31 @@ def _make_padding(schema: '_S', source: 'Any', name: 'str' = 'padding',
setattr(schema, name, bytes(padding))
return schema

def _make_options_trailer(self, block: 'Optional[Data_PCAPNG]',
options: 'list[Schema_Option | bytes]') -> 'int':
"""Append the octets a parsed block held past its ``opt_endofopt``.
def _make_options_trailer(self, block: 'Optional[Data_PCAPNG]') -> 'int':
"""Count the octets a parsed block held past its ``opt_endofopt``.

Args:
block: Block data model, if any.
options: Constructed options, extended in place.

Returns:
Number of octets appended.
Number of octets, which the block's length takes in.

An option area ends at its ``opt_endofopt`` and the octets after it are
read as padding. :meth:`_read_padding` keeps them as ``padding_opts``,
and they are written back after the options, verbatim (:issue:`1437`).
and :meth:`_make_padding` writes them back into the padding field after
the options, verbatim (:issue:`1437`). As an item of the options, they
would read as an area kept as captured, and a block rebuilt from its
``info`` would lose its options and ``padding_opts`` (:issue:`2776`).

.. versionchanged:: 1.6.0

Takes no ``options``, and no longer appends the octets to them
(:issue:`2776`).

"""
trailer = block.get('padding_opts') if block is not None else None
if not isinstance(trailer, bytes) or not trailer:
if not isinstance(trailer, (bytes, bytearray)) or not trailer:
return 0
options.append(trailer)
return len(trailer)

@classmethod
Expand Down
74 changes: 74 additions & 0 deletions tests/protocols/misc/test_pcapng_trailer_info_2776_unit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
# -*- coding: utf-8 -*-
"""GitHub issue #2776: a PCAP-NG block rebuilt from its ``info`` reads what its parse read.

Octets after ``opt_endofopt`` rebuilt byte for byte (#1437), but ``make`` wrote
them as an item of the options, and the block rebuilt from its ``info`` read the
options as an area kept as captured: no options, and no ``padding_opts``. They
are written into the block's padding field after the options instead, where the
parse reads them from.

Every case builds its own octets in memory and reads no capture.

"""

import warnings
from typing import TYPE_CHECKING

from tests.protocols.misc.test_pcapng_padding_kept_1437_unit import BLOCKS
from tests.protocols.misc.test_pcapng_roundtrip2_unit import PCAPNGTestCase

if TYPE_CHECKING:
from typing import Any

#: Test tier (#1958): hand-picked round-trip cases, not a sweep.
__tier__ = 'ALL'

#: The cases that hold octets after ``opt_endofopt``, one per block type with an option area.
TRAILED = {name: octets for name, octets in BLOCKS.items() if name.endswith('octets after opt_endofopt')}


class TestPCAPNGTrailerInfo(PCAPNGTestCase):
"""Pin ``from_data(info).info == info`` for every padding case of #1437."""

def _args(self, octets: 'bytes') -> 'dict[str, Any]':
"""The construction arguments ``_parse`` uses for ``octets``."""
if octets[:4] == b'\x0a\x0d\x0d\x0a':
return {'num': 0, 'sct': 1, 'ctx': None}
return {'num': 2, 'sct': 1, 'ctx': self._context()}

def test_every_block_type_with_an_option_area_has_a_case(self) -> None:
self.assertEqual(sorted(name.split()[0] for name in TRAILED),
['DSB', 'EPB', 'IDB', 'ISB', 'NRB', 'PB', 'SHB'])

def test_the_parse_reads_the_trailer_as_padding(self) -> None:
"""Against the octets: the four before the trailing length, after an ``opt_endofopt``."""
from pcapkit.const.pcapng.option_type import OptionType

for name, octets in TRAILED.items():
with self.subTest(name), warnings.catch_warnings():
warnings.simplefilter('ignore')
info = self._parse(octets).info
self.assertEqual(info.padding_opts, octets[-8:-4])
self.assertEqual(list(info.options)[-1], OptionType.opt_endofopt)

def test_info_rebuilds_as_parsed(self) -> None:
from pcapkit.protocols.misc.pcapng.protocol import PCAPNG

for name, octets in BLOCKS.items():
for form in ('info', 'to_dict'):
with self.subTest(name, form=form), warnings.catch_warnings():
warnings.simplefilter('ignore')
args = self._args(octets)
info = PCAPNG(octets, len(octets), **args).info
rebuilt = PCAPNG.from_data(info if form == 'info' else info.to_dict(), **args)
self.assertEqual(rebuilt.data.hex(), octets.hex())
self.assertEqual(rebuilt.info.get('padding_opts'), info.get('padding_opts'))
if 'options' in info:
self.assertEqual(list(rebuilt.info.options.items(multi=True)),
list(info.options.items(multi=True)))
self.assertEqual(rebuilt.info, info)


if __name__ == '__main__':
import unittest
unittest.main()
Loading