diff --git a/pcapkit/foundation/engines/pcapng.py b/pcapkit/foundation/engines/pcapng.py index 5c61bccd36..8c9b7b68c9 100644 --- a/pcapkit/foundation/engines/pcapng.py +++ b/pcapkit/foundation/engines/pcapng.py @@ -8,6 +8,7 @@ support, as is used by :class:`pcapkit.foundation.extraction.Extractor`. """ +import collections from typing import TYPE_CHECKING, cast from pcapkit.const.pcapng.block_type import BlockType as Enum_BlockType @@ -97,6 +98,9 @@ class PCAPNG(EngineBase[P_PCAPNG]): _ctx: 'Context' #: File context storage. _ctx_list: 'list[Context]' + #: Number of non-packet blocks of each kind written so far, counted + #: across the whole file rather than per section. + _block_count: 'collections.Counter[str]' MAGIC_NUMBER = ( b'\x0a\x0d\x0d\x0a', @@ -129,6 +133,7 @@ class PCAPNG(EngineBase[P_PCAPNG]): def __init__(self, extractor: 'Extractor') -> 'None': self._ctx = None # type: ignore[assignment] self._ctx_list = [] + self._block_count = collections.Counter() super().__init__(extractor) @@ -214,19 +219,19 @@ def read_frame(self) -> 'P_PCAPNG': elif block.info.type == Enum_BlockType.Interface_Description_Block: self._ctx.interfaces.append(cast('Data_InterfaceDescriptionBlock', block.info)) - self._write_file(block.info, name=f'Interface Description {len(self._ctx.interfaces)}') + self._write_file(block.info, name=self._block_name('Interface Description')) elif block.info.type == Enum_BlockType.Name_Resolution_Block: self._ctx.names.append(cast('Data_NameResolutionBlock', block.info)) - self._write_file(block.info, name=f'Name Resolution {len(self._ctx.names)}') + self._write_file(block.info, name=self._block_name('Name Resolution')) elif block.info.type == Enum_BlockType.systemd_Journal_Export_Block: self._ctx.journals.append(cast('Data_SystemdJournalExportBlock', block.info)) - self._write_file(block.info, name=f'systemd Journal Export {len(self._ctx.journals)}') + self._write_file(block.info, name=self._block_name('systemd Journal Export')) elif block.info.type == Enum_BlockType.Decryption_Secrets_Block: self._ctx.secrets.append(cast('Data_DecryptionSecretsBlock', block.info)) - self._write_file(block.info, name=f'Decryption Secrets {len(self._ctx.secrets)}') + self._write_file(block.info, name=self._block_name('Decryption Secrets')) elif block.info.type == Enum_BlockType.Interface_Statistics_Block: # NOTE: The interface ID is bounds-checked while the block is @@ -237,12 +242,12 @@ def read_frame(self) -> 'P_PCAPNG': isb_info = cast('Data_InterfaceStatisticsBlock', block.info) self._ctx.statistics.append(isb_info) - self._write_file(isb_info, name=f'Interface Statistics {len(self._ctx.statistics)}') + self._write_file(isb_info, name=self._block_name('Interface Statistics')) elif block.info.type in (Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files, Enum_BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files): self._ctx.custom.append(cast('Data_CustomBlock', block.info)) - self._write_file(block.info, name=f'Custom {len(self._ctx.custom)}') + self._write_file(block.info, name=self._block_name('Custom')) elif block.info.type == Enum_BlockType.Enhanced_Packet_Block: # NOTE: as for the ISB above, the interface ID has already been @@ -267,7 +272,7 @@ def read_frame(self) -> 'P_PCAPNG': else: self._ctx.unknown.append(cast('Data_UnknownBlock', block.info)) - self._write_file(block.info, name=f'Unknown {len(self._ctx.unknown)}') + self._write_file(block.info, name=self._block_name('Unknown')) # increment frame number ext._frnum += 1 @@ -311,6 +316,26 @@ def read_frame(self) -> 'P_PCAPNG': # Utilities. ########################################################################## + def _block_name(self, kind: 'str') -> 'str': + """Name the next non-packet block of the given kind. + + The per-section context lists (e.g. ``Context.interfaces``) restart + with every Section Header Block, so numbering a block by their length + gives the first IDB of every section the same name. The output dumpers + key by name, so a repeated name overwrites the earlier block (#1256). + The number is therefore counted across the whole file instead; for a + single-section file it is the same number as before. + + Args: + kind: The kind of the block, e.g. ``'Interface Description'``. + + Returns: + The block name, ``' '`` with ``n`` starting at 1. + + """ + self._block_count[kind] += 1 + return f'{kind} {self._block_count[kind]}' + def _write_file(self, block: 'Data_PCAPNG', *, name: 'str') -> 'None': """Write the parsed block into output file. diff --git a/tests/foundation/test_pcapng_unique_block_names_unit.py b/tests/foundation/test_pcapng_unique_block_names_unit.py new file mode 100644 index 0000000000..e8f60318a0 --- /dev/null +++ b/tests/foundation/test_pcapng_unique_block_names_unit.py @@ -0,0 +1,118 @@ +# -*- coding: utf-8 -*- +"""Regression tests for #1256: PCAP-NG block names are unique across sections. + +The PCAP-NG engine used to number a non-packet block by the length of its +section's context list, which restarts with every Section Header Block. The +first Interface Description Block of the second section was therefore named +``Interface Description 1`` again, so a mapping dumper (JSON, plist) kept only +one of the two blocks and ``files=True`` overwrote the first block's file. + +""" +from __future__ import annotations + +import json +import os +import struct +import tempfile +import unittest +import warnings + +from pcapkit.interface import extract +from tests._support import sample_path +from tests.foundation.engines.test_pcapng_engine import PCAPNGWriter + +#: Every non-packet block kind the engine names by a per-kind counter. +NAMED_KINDS = ('Interface Description', 'Name Resolution', 'systemd Journal Export', + 'Decryption Secrets', 'Interface Statistics', 'Custom', 'Unknown') + + +class SectionWriter(PCAPNGWriter): + """:class:`PCAPNGWriter` plus the remaining non-packet block kinds.""" + + def name_resolution(self) -> SectionWriter: + """Name Resolution Block holding only the end-of-records record.""" + return self._block(0x00000004, struct.pack(f'{self._endian}HH', 0, 0)) + + def journal_export(self) -> SectionWriter: + """systemd Journal Export Block with one journal entry field.""" + return self._block(0x00000009, b'MESSAGE=1256\n') + + def decryption_secrets(self) -> SectionWriter: + """Decryption Secrets Block carrying a TLS key log line.""" + secret = b'CLIENT_RANDOM 00 00\n' + return self._block(0x0000000A, struct.pack(f'{self._endian}II', 0x544c534b, len(secret)) + + self._pad(secret)) + + def custom(self) -> SectionWriter: + """Custom Block (copyable) with a private enterprise number and no data.""" + return self._block(0x00000BAD, struct.pack(f'{self._endian}I', 32473)) + + def unknown(self) -> SectionWriter: + """A reserved block type, which the engine records as an unknown block.""" + return self._block(0x00000000, b'\x00' * 4) + + def section(self) -> SectionWriter: + """One section holding every named block kind once, then one packet.""" + self.section_header().interface_description().name_resolution() + self.journal_export().decryption_secrets().interface_statistics() + self.custom().unknown().enhanced_packet() + return self + + +def dumped(path: str) -> tuple[list[str], list[str]]: + """Extract ``path`` to JSON twice: as one file, and as one file per block. + + Returns: + The top-level keys of the JSON file in order, duplicates included, and + the sorted file names of the ``files=True`` output directory. + + """ + with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(): + warnings.simplefilter('ignore') + out = os.path.join(tmp, 'out.json') + extract(fin=path, fout=out, format='json', store=False, extension=False, + engine='pcapkit') + with open(out, encoding='utf-8') as file: + keys = [key for key, _ in json.load(file, object_pairs_hook=lambda pairs: pairs)] + + folder = os.path.join(tmp, 'blocks') + extract(fin=path, fout=folder, format='json', store=False, extension=False, + files=True, engine='pcapkit') + files = sorted(os.listdir(folder)) + return keys, files + + +class SingleSectionSampleTests(unittest.TestCase): + """A single-section file is named exactly as before the fix.""" + + def test_single_section_names_are_unchanged(self) -> None: + keys, files = dumped(sample_path('dhcp.pcapng')) + self.assertEqual(keys, ['Section Header 1', 'Interface Description 1', + 'Frame 1', 'Frame 2', 'Frame 3', 'Frame 4']) + self.assertEqual(files, sorted(f'{key}.json' for key in keys)) + + +class EveryKindAcrossSectionsTests(unittest.TestCase): + """Each per-kind counter runs on across sections, not just the IDB one.""" + + def test_second_section_continues_every_counter(self) -> None: + capture = SectionWriter().section().section() + handle, path = tempfile.mkstemp(suffix='.pcapng') + try: + with os.fdopen(handle, 'wb') as file: + file.write(bytes(capture)) + keys, files = dumped(path) + finally: + os.unlink(path) + + expected = [] + for number in (1, 2): + expected.append(f'Section Header {number}') + expected.extend(f'{kind} {number}' for kind in NAMED_KINDS) + expected.append(f'Frame {number}') + self.assertEqual(keys, expected) + self.assertEqual(files, sorted(f'{key}.json' for key in expected)) + + +if __name__ == '__main__': + unittest.main()