Skip to content

fix(pcapng): constructed and rebuilt blocks round-trip (#1267, #1268, #1269, #1270, #1271) #1339

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
Oct 7, 2026
+315 −84
Merged

fix(pcapng): constructed and rebuilt blocks round-trip (#1267, #1268, #1269, #1270, #1271) #1339

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
8 changes: 8 additions & 0 deletions pcapkit/protocols/data/misc/pcapng.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -769,6 +769,10 @@ class TLSKeyLog(DSBSecrets):
entries: 'dict[TLSKeyLabel, OrderedMultiDict[bytes, bytes]]'

if TYPE_CHECKING:
#: TLS key log text as the block carries it, comments and line order
#: included. Set on parsing only, and written back verbatim on rebuild.
data: 'str'

def __init__(self, entries: 'dict[TLSKeyLabel, OrderedMultiDict[bytes, bytes]]') -> 'None': ...


Expand All @@ -780,6 +784,10 @@ class WireGuardKeyLog(DSBSecrets):
entries: 'OrderedMultiDict[WireGuardKeyLabel, bytes]'

if TYPE_CHECKING:
#: WireGuard key log text as the block carries it, comments included.
#: Set on parsing only, and written back verbatim on rebuild.
data: 'str'

def __init__(self, entries: 'OrderedMultiDict[WireGuardKeyLabel, bytes]') -> 'None': ...


Expand Down
79 changes: 71 additions & 8 deletions pcapkit/protocols/misc/pcapng.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
import enum
import io
import math
import os
import platform
import re
import struct
Expand Down Expand Up @@ -1237,6 +1236,12 @@ def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Opt
self._data = self.pack(**kwargs)
#: io.BytesIO: Source packet stream.
self._file = io.BytesIO(self._data)

# NOTE: The block just built is parsed again below on this same
# instance, and the parse counts each option afresh against its
# "only one" guard -- so the count left by the make pass must not
# carry over into it.
self._opt = collections.Counter()
else:
_read = True
#: io.BytesIO: Source packet stream.
Expand Down Expand Up @@ -1572,6 +1577,34 @@ def _make_data(cls, data: 'Data_PCAPNG') -> 'dict[str, Any]': # type: ignore[ov
'block': data,
}

@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.

Args:
block: Block data model.

Returns:
The captured packet octets the block was parsed from, as
:attr:`~pcapkit.protocols.data.protocol.Protocol.packet` holds them,
cut to the block's ``captured_len``; or, for a data model that
carries none, the payload rebuilt from its decoded next layer.

Note:
The cut is what keeps a block with ``captured_len`` of zero intact:
its ``packet`` can hold octets past the payload (:issue:`1275`), and
writing those back would grow the block.

The fallback is lossy. A next layer rebuilt from its own data model
drops anything it did not decode, such as Ethernet pad octets, so
the block it yields can be shorter than its ``captured_len`` says.

"""
packet = block.get('packet')
if isinstance(packet, bytes):
return packet[:block.captured_len]
return cls._make_payload(block)

def _make_timestamp(self, timestamp: 'Optional[float | Decimal | dt_type | int]' = None, *,
interface_id: 'int' = 0) -> 'tuple[int, int]':
"""Make timestamp.
Expand Down Expand Up @@ -3492,6 +3525,10 @@ def _read_secrets_tls(self, schema: 'Schema_TLSKeyLog', *,
secrets = Data_TLSKeyLog(
entries=schema.entries,
)
# NOTE: the log text is kept as well, since :attr:`entries` holds
# neither its comments nor the order of its lines, and a rebuild writes
# it back verbatim.
secrets.__update__(data=schema.data)
return secrets

def _read_secrets_wireguard(self, schema: 'Schema_WireGuardKeyLog', *,
Expand All @@ -3509,6 +3546,10 @@ def _read_secrets_wireguard(self, schema: 'Schema_WireGuardKeyLog', *,
secrets = Data_WireGuardKeyLog(
entries=schema.entries,
)
# NOTE: the log text is kept as well, since :attr:`entries` holds
# neither its comments nor the order of its lines, and a rebuild writes
# it back verbatim.
secrets.__update__(data=schema.data)
return secrets

def _read_secrets_zigbee_nwk(self, schema: 'Schema_ZigBeeNWKKey', *,
Expand Down Expand Up @@ -3757,6 +3798,7 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *,
timestamp = block.timestamp_epoch
captured_len = block.captured_len
original_len = block.original_len
packet_data = self._make_packet_data(block)
options = block.options

timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id)
Expand Down Expand Up @@ -3805,6 +3847,7 @@ def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *,
"""
if block is not None:
original_len = block.original_len
packet_data = self._make_packet_data(block)

if original_len is None:
original_len = len(packet_data)
Expand Down Expand Up @@ -3878,6 +3921,9 @@ def _make_block_isb(self, block: 'Optional[Data_InterfaceStatisticsBlock]' = Non
options = block.options

timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id)
# NOTE: ``isb_starttime`` and ``isb_endtime`` are scaled by the
# resolution of this block's interface, as on the parse path.
self._isb_interface_id = interface_id # pylint: disable=attribute-defined-outside-init

if options is not None:
options_value, total_length = self._make_pcapng_options(options, namespace='isb')
Expand Down Expand Up @@ -4094,6 +4140,7 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *,
timestamp = block.timestamp_epoch
captured_len = block.captured_len
original_len = block.original_len
packet_data = self._make_packet_data(block)
options = block.options

timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id)
Expand Down Expand Up @@ -4427,7 +4474,7 @@ def _make_option_if_ipv6(self, type: 'Enum_OptionType', option: 'Optional[Data_I

return Schema_IF_IPv6AddrOption(
type=type,
length=8,
length=17,
interface=interface,
)

Expand Down Expand Up @@ -5774,7 +5821,10 @@ def _make_secrets_tls(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_TL

Args:
type: Secrets type.
secrets: Secrets data model.
secrets: Secrets data model. The key log text it was parsed from,
if any, is written back verbatim; otherwise its entries are.
On a parsed model the text wins, so an edit to its
``entries`` alone is silently ignored.
entries: TLS Key Log entries.
**kwargs: Arbitrary keyword arguments.

Expand All @@ -5785,15 +5835,20 @@ def _make_secrets_tls(self, type: 'Enum_SecretsType', secrets: 'Optional[Data_TL
from pcapkit import __version__

if secrets is not None:
text = secrets.get('data')
if isinstance(text, str):
return Schema_TLSKeyLog(
data=text,
)
entries = secrets.entries

if entries is None:
entries = {}

data = [f'# generated by PyPCAPKit v{__version__} at {datetime.datetime.now().isoformat()}{os.sep}'] # type: list[str]
data = [f'# generated by PyPCAPKit v{__version__}\n'] # type: list[str]
for label, entry in entries.items():
for k, v in entry.items(multi=True):
data.append(f'{label.name} {k.hex()} {v.hex()}{os.sep}')
data.append(f'{label.name} {k.hex()} {v.hex()}\n')

return Schema_TLSKeyLog(
data=''.join(data),
Expand All @@ -5806,7 +5861,10 @@ def _make_secrets_wireguard(self, type: 'Enum_SecretsType', secrets: 'Optional[D

Args:
type: Secrets type.
secrets: Secrets data model.
secrets: Secrets data model. The key log text it was parsed from,
if any, is written back verbatim; otherwise its entries are.
On a parsed model the text wins, so an edit to its
``entries`` alone is silently ignored.
entries: WireGuard Key Log entries.
**kwargs: Arbitrary keyword arguments.

Expand All @@ -5817,14 +5875,19 @@ def _make_secrets_wireguard(self, type: 'Enum_SecretsType', secrets: 'Optional[D
from pcapkit import __version__

if secrets is not None:
text = secrets.get('data')
if isinstance(text, str):
return Schema_WireGuardKeyLog(
data=text,
)
entries = secrets.entries

if entries is None:
entries = OrderedMultiDict()

data = [f'# generated by PyPCAPKit v{__version__} at {datetime.datetime.now().isoformat()}{os.sep}'] # type: list[str]
data = [f'# generated by PyPCAPKit v{__version__}\n'] # type: list[str]
for label, value in entries.items(multi=True):
data.append(f'{label.name} = {base64.b64encode(value).decode()}{os.sep}')
data.append(f'{label.name} = {base64.b64encode(value).decode()}\n')

return Schema_WireGuardKeyLog(
data=''.join(data),
Expand Down
Loading
Loading