fix(pcapng): rebuilt blocks keep their lengths and padding (#1276, #1279, #1280, #1282, #1283, #1284)

pcapkit/protocols/misc/pcap/frame.py

@@ -367,6 +367,11 @@ def make(self,
             incl_len = min(len(packet), self._ghdr.snaplen)
         if orig_len is None:
             orig_len = len(packet)
+        # NOTE: The record holds exactly ``incl_len`` octets of packet data, so a
+        # packet longer than that -- e.g. than the snaplen -- is cut to it rather
+        # than written out whole behind a clipped length.
+        if len(packet) > incl_len:
+            packet = bytes(packet)[:incl_len]
 
         return Schema_Frame(
             ts_sec=ts_sec,

pcapkit/protocols/misc/pcapng.py

@@ -38,6 +38,7 @@
 from pcapkit.const.pcapng.verdict_type import VerdictType as Enum_VerdictType
 from pcapkit.const.reg.linktype import LinkType as Enum_LinkType
 from pcapkit.corekit.enum import EnumLookup
+from pcapkit.corekit.fields.strings import DecodedString
 from pcapkit.corekit.module import ModuleDescriptor
 from pcapkit.corekit.multidict import OrderedMultiDict
 from pcapkit.corekit.version import VersionInfo
@@ -245,6 +246,22 @@ def _option_key(code: 'Enum_OptionType') -> 'Union[Enum_OptionType, Tuple[str, i
     return code
 
 
+def _utf8_length(text: 'str') -> 'int':
+    """Return the number of octets a UTF-8 :class:`~pcapkit.corekit.fields.strings.StringField` packs ``text`` as.
+
+    Args:
+        text: String value; a :class:`~pcapkit.corekit.fields.strings.DecodedString`
+            packs as the octets it was decoded from.
+
+    Returns:
+        Length of the encoded value, in octets.
+
+    """
+    if isinstance(text, DecodedString):
+        return len(text.raw)
+    return len(text.encode('utf-8'))
+
+
 class PacketDirection(EnumLookup, enum.IntEnum):
     """Packet direction for ``epb_flags`` options.
 
@@ -1033,6 +1050,12 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_PCAPN
             if self._ctx is not None:
                 self._byte = self._ctx.section.byteorder
                 packet['byteorder'] = self._byte
+                # NOTE: A Simple Packet Block's data is ``min(original_len,
+                # snaplen)`` octets, with the snaplen of interface 0.
+                # A snaplen of zero means no limit.
+                interfaces = getattr(self._ctx, 'interfaces', None)
+                if interfaces and interfaces[0].snaplen:
+                    packet.setdefault('snaplen', interfaces[0].snaplen)
             self.__header__ = cast('Schema_PCAPNG', self.__schema__.unpack(self._file, length, packet))  # type: ignore[call-arg,misc]
 
         data = self.read(length, **kwargs)
@@ -2422,7 +2445,14 @@ def _read_option_custom(self, schema: 'Schema_CustomOption', *,
         Returns:
             Constructed option data.
 
+        Raises:
+            ProtocolError: If the option is shorter than its 4-octet private
+                enterprise number, which Option Length includes.
+
         """
+        if schema.length < 4:
+            raise ProtocolError(f'PCAP-NG: [opt_custom] invalid length (expected at least 4, got {schema.length})')
+
         option = Data_CustomOption(
             type=schema.type,
             length=schema.length,
@@ -3806,10 +3836,16 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *,
             if self._ctx is None:
                 snaplen = 0xFFFF_FFFF_FFFF_FFFF
             else:
-                snaplen = self._get_interface(interface_id).snaplen
+                # a snaplen of zero means no limit
+                snaplen = self._get_interface(interface_id).snaplen or 0xFFFF_FFFF_FFFF_FFFF
             captured_len = min(len(packet_data), snaplen)
         if original_len is None:
             original_len = len(packet_data)
+        # NOTE: Packet Data holds exactly Captured Packet Length octets, so a
+        # packet longer than that -- e.g. than the interface's snaplen -- is cut
+        # to it rather than written out whole behind a clipped length.
+        if len(packet_data) > captured_len:
+            packet_data = bytes(packet_data)[:captured_len]
         packet_len = math.ceil(len(packet_data) / 4) * 4
 
         if options is not None:
@@ -3851,6 +3887,16 @@ def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *,
 
         if original_len is None:
             original_len = len(packet_data)
+        # NOTE: An SPB always belongs to interface 0, and its Packet Data is
+        # ``min(original_len, snaplen)`` octets long -- the length the parser
+        # reads back -- so a longer packet is cut to it.
+        if self._ctx is None or not self._ctx.interfaces:
+            snaplen = 0xFFFF_FFFF_FFFF_FFFF
+        else:
+            snaplen = self._ctx.interfaces[0].snaplen or 0xFFFF_FFFF_FFFF_FFFF
+        captured_len = min(original_len, snaplen)
+        if len(packet_data) > captured_len:
+            packet_data = bytes(packet_data)[:captured_len]
         packet_len = math.ceil(len(packet_data) / 4) * 4
 
         return Schema_SimplePacketBlock(
@@ -3980,6 +4026,9 @@ def _make_block_systemd(self, block: 'Optional[Data_SystemdJournalExportBlock]'
 
                 temp.append(b''.join(tmp_buf))
             data = b'\n'.join(temp)
+        # NOTE: The entry MUST be zero-padded to 32 bits, and the Block Total
+        # Length counts that padding, as the ``entry`` field reads it back.
+        data += bytes(-len(data) % 4)
 
         return Schema_SystemdJournalExportBlock(
             length=len(data) + 12,
@@ -4038,20 +4087,23 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None,
             secrets_data_val = secrets_data
         else:
             raise ProtocolError(f'PCAP-NG: [DSB] secrets {secrets_type_val} invalid format')
+        # NOTE: Secrets Length excludes the padding to the next 32-bit boundary,
+        # which ``padding_data`` writes and the block length has to include.
         secrets_length = len(secrets_data_val)
+        secrets_padded = secrets_length + -secrets_length % 4
 
         if options is not None:
             options_value, total_length = self._make_pcapng_options(options, namespace='dsb')
         else:
             options_value, total_length = [], 0
 
         return Schema_DecryptionSecretsBlock(
-            length=total_length + secrets_length + 20,
+            length=total_length + secrets_padded + 20,
             secrets_type=secrets_type_val,
             secrets_length=secrets_length,
             secrets_data=secrets_data_val,
             options=options_value,
-            length2=total_length + secrets_length + 20,
+            length2=total_length + secrets_padded + 20,
         )
 
     def _make_block_cb(self, block: 'Optional[Data_CustomBlock]' = None, *,
@@ -4148,10 +4200,16 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *,
             if self._ctx is None:
                 snaplen = 0xFFFF_FFFF_FFFF_FFFF
             else:
-                snaplen = self._get_interface(interface_id).snaplen
+                # a snaplen of zero means no limit
+                snaplen = self._get_interface(interface_id).snaplen or 0xFFFF_FFFF_FFFF_FFFF
             captured_len = min(len(packet_data), snaplen)
         if original_len is None:
             original_len = len(packet_data)
+        # NOTE: Packet Data holds exactly Captured Packet Length octets, so a
+        # packet longer than that -- e.g. than the interface's snaplen -- is cut
+        # to it rather than written out whole behind a clipped length.
+        if len(packet_data) > captured_len:
+            packet_data = bytes(packet_data)[:captured_len]
         packet_len = math.ceil(len(packet_data) / 4) * 4
 
         if options is not None:
@@ -4328,7 +4386,7 @@ def _make_option_comment(self, type: 'Enum_OptionType', option: 'Optional[Data_C
 
         return Schema_CommentOption(
             type=type,
-            length=len(comment),
+            length=_utf8_length(comment),
             comment=comment,
         )
 
@@ -4387,7 +4445,7 @@ def _make_option_if_name(self, type: 'Enum_OptionType', option: 'Optional[Data_I
 
         return Schema_IF_NameOption(
             type=type,
-            length=len(name),
+            length=_utf8_length(name),
             name=name,
         )
 
@@ -4418,7 +4476,7 @@ def _make_option_if_description(self, type: 'Enum_OptionType', option: 'Optional
 
         return Schema_IF_DescriptionOption(
             type=type,
-            length=len(description),
+            length=_utf8_length(description),
             description=description,
         )
 
@@ -4731,7 +4789,7 @@ def _make_option_if_os(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_
 
         return Schema_IF_OSOption(
             type=type,
-            length=len(os),
+            length=_utf8_length(os),
             os=os,
         )
 
@@ -4824,7 +4882,7 @@ def _make_option_if_hardware(self, type: 'Enum_OptionType', option: 'Optional[Da
 
         return Schema_IF_HardwareOption(
             type=type,
-            length=len(hardware),
+            length=_utf8_length(hardware),
             hardware=hardware,
         )
 
@@ -5185,7 +5243,7 @@ def _make_option_ns_dnsname(self, type: 'Enum_OptionType', option: 'Optional[Dat
 
         return Schema_NS_DNSNameOption(
             type=type,
-            length=len(name),
+            length=_utf8_length(name),
             name=name,
         )
 

pcapkit/protocols/schema/misc/pcapng.py

@@ -1408,9 +1408,11 @@ class SimplePacketBlock(BlockType, code=Enum_BlockType.Simple_Packet_Block):
     length: 'int' = UInt32Field(callback=byteorder_callback)
     #: Original packet length.
     original_len: 'int' = UInt32Field(callback=byteorder_callback)
-    #: Packet data.
-    packet_data: 'bytes' = PayloadField(length=lambda pkt: min(pkt.get('snaplen', 0xFFFFFFFFFFFFFFFF),
-                                                               pkt['original_len']))
+    #: Packet data, ``min(original_len, snaplen)`` octets; a packet cut by the
+    #: snaplen is bounded by what the block holds, less its 16 framing octets.
+    packet_data: 'bytes' = PayloadField(length=lambda pkt: max(0, min(pkt.get('snaplen', 0xFFFFFFFFFFFFFFFF),
+                                                                      pkt['original_len'],
+                                                                      pkt['length'] - 16)))
     #: Padding.
     padding: 'bytes' = PaddingField(length=lambda pkt: (4 - len(pkt['packet_data']) % 4) % 4)
     #: Block total length.
@@ -2105,8 +2107,10 @@ class ZigBeeNWKKey(DSBSecrets, code=Enum_SecretsType.ZigBee_NWK_Key):
     key: 'bytes' = BytesField(length=16)
     #: ZigBee PANID.
     panid: 'int' = UInt16Field(byteorder='little')
-    #: Padding.
-    padding: 'bytes' = BytesField(length=2)
+
+    # NOTE: The two zero octets after the PAN ID are the Decryption Secrets
+    # Block's own padding, which Secrets Length does not count, so they belong
+    # to :attr:`DecryptionSecretsBlock.padding_data` rather than to this schema.
 
     if TYPE_CHECKING:
         def __init__(self, key: 'bytes', panid: 'int') -> 'None': ...
@@ -2124,8 +2128,9 @@ class ZigBeeAPSKey(DSBSecrets, code=Enum_SecretsType.ZigBee_APS_Key):
     addr_low: 'int' = UInt16Field(byteorder='little')
     #: High node short address.
     addr_high: 'int' = UInt16Field(byteorder='little')
-    #: Padding.
-    padding: 'bytes' = BytesField(length=2)
+
+    # NOTE: As for :class:`ZigBeeNWKKey`, the trailing two zero octets are the
+    # block's padding, not part of the secrets.
 
     if TYPE_CHECKING:
         def __init__(self, key: 'bytes', panid: 'int', addr_low: 'int', addr_high: 'int') -> 'None': ...