fix(foundation): number pcapng blocks across the whole file, not per section (#1256)

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, ``'<kind> <n>'`` 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.
 

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()