fix(sctp): rebuild a last chunk short of its padding as captured (#1474)

pcapkit/protocols/data/transport/sctp.py

@@ -104,7 +104,11 @@ class ErrorCause(Data):
 
     if TYPE_CHECKING:
         #: Trailing padding octets, as captured; present only when they are
-        #: not all zeros.
+        #: not all zeros, or fewer than the ``length`` calls for (:issue:`1474`).
+        #: Read back by ``from_data`` only, where fewer octets are honoured on
+        #: the last item of its list alone and zero-filled to the full width
+        #: anywhere else. ``make`` builds from keywords and takes no
+        #: ``padding``: every item it builds is padded in zeros.
         padding: 'bytes'
 
 
@@ -267,7 +271,11 @@ class Parameter(Data):
 
     if TYPE_CHECKING:
         #: Trailing padding octets, as captured; present only when they are
-        #: not all zeros.
+        #: not all zeros, or fewer than the ``length`` calls for (:issue:`1474`).
+        #: Read back by ``from_data`` only, where fewer octets are honoured on
+        #: the last item of its list alone and zero-filled to the full width
+        #: anywhere else. ``make`` builds from keywords and takes no
+        #: ``padding``: every item it builds is padded in zeros.
         padding: 'bytes'
 
 
@@ -380,7 +388,11 @@ class Chunk(Data):
 
     if TYPE_CHECKING:
         #: Trailing padding octets, as captured; present only when they are
-        #: not all zeros.
+        #: not all zeros, or fewer than the ``length`` calls for (:issue:`1474`).
+        #: Read back by ``from_data`` only, where fewer octets are honoured on
+        #: the last item of its list alone and zero-filled to the full width
+        #: anywhere else. ``make`` builds from keywords and takes no
+        #: ``padding``: every item it builds is padded in zeros.
         padding: 'bytes'
 
 

pcapkit/protocols/schema/transport/sctp.py

@@ -96,18 +96,45 @@ def padding_length(pkt: 'dict[str, Any]') -> 'int':
     Returns:
         Number of padding bytes, clamped to the number of bytes left in the
         enclosing structure, since :rfc:`9260#section-3.2` allows the final
-        padding of a packet to be omitted.
+        padding of a packet to be omitted. When packing, it is clamped as well
+        to the ``__padding__`` width :func:`pre_pack_padding` hands over, so a
+        final chunk captured without (all of) its padding rebuilds without it
+        (:issue:`1474`).
 
     """
     length = pkt.get('length') or 0
     padding = -length % 4
 
+    width = pkt.get('__padding__')
+    if isinstance(width, int) and width >= 0:
+        padding = min(padding, width)
+
     remaining = pkt.get('__length__')
     if not isinstance(remaining, int) or remaining < 0:
         return padding
     return min(padding, remaining)
 
 
+def pre_pack_padding(schema: 'Schema', packet: 'dict[str, Any]') -> 'None':
+    """Hand a schema's recorded padding width to :func:`padding_length`.
+
+    Args:
+        schema: Chunk, parameter or error cause being packed.
+        packet: Packet data, shared with the enclosing and sibling schemas.
+
+    A schema whose padding was captured short -- only the last chunk of a
+    packet can be -- carries the width as captured under ``__padding__`` in its
+    ``__dict__``, set by :func:`~pcapkit.protocols.transport.sctp._restore_padding`
+    from the chunk's ``padding`` as parsed (:issue:`1474`). The key is written to
+    ``packet`` for *every* schema, ``None`` when there is no record, since
+    :meth:`Schema.pack <pcapkit.protocols.schema.schema.Schema.pack>` shares one
+    ``packet`` mapping across siblings and a record left behind by one would
+    otherwise shorten the padding of the next.
+
+    """
+    packet['__padding__'] = schema.__dict__.get('__padding__')
+
+
 def bounded(length: 'Callable[[dict[str, Any]], int]') -> 'Callable[[dict[str, Any]], int]':
     """Clamp a list field's computed length to the bytes actually available.
 
@@ -261,6 +288,10 @@ class ErrorCause(EnumSchema[Enum_CauseCode]):
     #: Cause length.
     length: 'int' = UInt16Field()
 
+    def pre_pack(self, packet: 'dict[str, Any]') -> 'None':
+        """Prepare ``packet`` data for packing process; see :func:`pre_pack_padding`."""
+        pre_pack_padding(self, packet)
+
 
 @schema_final
 class UnknownCause(ErrorCause):
@@ -457,6 +488,10 @@ class Parameter(EnumSchema[Enum_Parameter]):
     #: Parameter length.
     length: 'int' = UInt16Field()
 
+    def pre_pack(self, packet: 'dict[str, Any]') -> 'None':
+        """Prepare ``packet`` data for packing process; see :func:`pre_pack_padding`."""
+        pre_pack_padding(self, packet)
+
 
 @schema_final
 class UnknownParameter(Parameter):
@@ -606,6 +641,10 @@ class Chunk(EnumSchema[Enum_Chunk]):
     #: Chunk length, excluding any trailing padding.
     length: 'int' = UInt16Field()
 
+    def pre_pack(self, packet: 'dict[str, Any]') -> 'None':
+        """Prepare ``packet`` data for packing process; see :func:`pre_pack_padding`."""
+        pre_pack_padding(self, packet)
+
 
 @schema_final
 class UnknownChunk(Chunk):

pcapkit/protocols/transport/sctp.py

@@ -182,6 +182,7 @@
 
     Data_T = TypeVar('Data_T', bound=Data)
     Schema_T = TypeVar('Schema_T', bound=Schema)
+    List_T = TypeVar('List_T', bound=list)
 
     ChunkParser = Callable[[Schema_Chunk, NamedArg(Chunks, 'chunks')], Data_Chunk]
     ChunkConstructor = Callable[[Enum_Chunk, DefaultArg(Optional[Data_Chunk]),
@@ -218,16 +219,34 @@ def _keep_padding(schema: 'Schema', data: 'Data_T') -> 'Data_T':
 
     Returns:
         ``data``, with a ``padding`` attribute holding the trailing padding as
-        captured if it is not all zeros -- the form a fresh build writes
-        (:issue:`1223`). :func:`_restore_padding` writes it back.
+        captured if it is not what a fresh build writes: all zeros
+        (:issue:`1223`), and as many as the ``length`` calls for. A final chunk
+        may arrive with fewer, or none (:issue:`1474`), in which case
+        ``padding`` holds the octets that were there, possibly ``b''``.
+        :func:`_restore_padding` writes it back.
 
     """
     padding = schema.__dict__.get('padding')
-    if isinstance(padding, bytes) and any(padding):
+    if isinstance(padding, bytes) and (any(padding) or len(padding) < _padding_width(schema)):
         data.__update__([('padding', padding)])
     return data
 
 
+def _padding_width(schema: 'Schema') -> 'int':
+    """Number of padding octets a fresh build writes after ``schema``.
+
+    Args:
+        schema: Chunk, parameter or error cause schema.
+
+    Returns:
+        The octets that round its ``length`` up to a multiple of four
+        (:rfc:`9260#section-3.2`).
+
+    """
+    length = schema.__dict__.get('length')
+    return -length % 4 if isinstance(length, int) else 0
+
+
 def _restore_padding(data: 'Optional[Data]', schema: 'Schema_T') -> 'Schema_T':
     """Write back the trailing padding :func:`_keep_padding` kept, if any.
 
@@ -236,15 +255,60 @@ def _restore_padding(data: 'Optional[Data]', schema: 'Schema_T') -> 'Schema_T':
         schema: Constructed schema.
 
     Returns:
-        ``schema``, with its padding set from ``data``.
+        ``schema``, with its padding set from ``data``. Padding kept shorter
+        than ``length`` calls for also sets the schema's ``__padding__`` to its
+        width, which :func:`~pcapkit.protocols.schema.transport.sctp.padding_length`
+        packs to rather than zero-filling the octets that were never captured
+        (:issue:`1474`).
 
     """
     padding = getattr(data, 'padding', None)
     if isinstance(padding, bytes) and 'padding' in schema.__fields__:
         schema.padding = padding
+        if len(padding) < _padding_width(schema):
+            schema.__dict__['__padding__'] = len(padding)
     return schema
 
 
+def _last_item_only(schemas: 'List_T') -> 'List_T':
+    """Drop the short-padding record from every item but the last.
+
+    Args:
+        schemas: Constructed chunk, parameter or cause list.
+
+    Returns:
+        ``schemas``, in which only the last item may keep the ``__padding__``
+        width :func:`_restore_padding` set. Only the end of a packet may be
+        short of its padding (:issue:`1474`); a short record on any earlier
+        item -- from an edited ``info``, say -- would misalign everything after
+        it, so that item packs to the full width, its ``padding`` octets
+        zero-filled to it. The same goes for the parameters or causes nested in
+        an earlier chunk, whose last item ends that chunk rather than the packet.
+
+    """
+    for schema in schemas[:-1]:
+        _drop_padding_record(schema)
+    return schemas
+
+
+def _drop_padding_record(schema: 'Schema | bytes') -> 'None':
+    """Drop ``schema``'s short-padding record, and its nested items' too.
+
+    Args:
+        schema: Chunk, parameter or cause that does not end the packet.
+
+    """
+    if not isinstance(schema, Schema):
+        return
+    if schema.__dict__.pop('__padding__', None) is not None:
+        schema.__updated__ = True
+    for name in ('parameters', 'error'):
+        nested = schema.__dict__.get(name)
+        if isinstance(nested, list):
+            for item in nested:
+                _drop_padding_record(item)
+
+
 class SCTP(Transport[Data_SCTP, Schema_SCTP],
            schema=Schema_SCTP, data=Data_SCTP):
     """This class implements Stream Control Transmission Protocol.
@@ -987,11 +1051,11 @@ def _make_sctp_chunks(self, chunks: 'list[Schema_Chunk | tuple[Enum_Chunk, dict[
                 else:
                     code, args = cast('tuple[Enum_Chunk, dict[str, Any]]', schema)
                     chunks_list.append(self._make_sctp_chunk(code, None, **args))
-            return chunks_list
+            return _last_item_only(chunks_list)
 
         for code, chunk in chunks.items(multi=True):
             chunks_list.append(self._make_sctp_chunk(code, chunk))
-        return chunks_list
+        return _last_item_only(chunks_list)
 
     def _make_sctp_chunk(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Chunk]' = None,
                          **kwargs: 'Any') -> 'Schema_Chunk':
@@ -1059,11 +1123,11 @@ def _make_sctp_parameters(self, parameters: 'list[Schema_Parameter | tuple[Enum_
                 else:
                     code, args = cast('tuple[Enum_Parameter, dict[str, Any]]', schema)
                     parameters_list.append(self._make_sctp_parameter(code, None, **args))
-            return parameters_list
+            return _last_item_only(parameters_list)
 
         for code, parameter in parameters.items(multi=True):
             parameters_list.append(self._make_sctp_parameter(code, parameter))
-        return parameters_list
+        return _last_item_only(parameters_list)
 
     def _make_sctp_parameter(self, code: 'Enum_Parameter', parameter: 'Optional[Data_Parameter]' = None,
                              **kwargs: 'Any') -> 'Schema_Parameter':
@@ -1131,11 +1195,11 @@ def _make_sctp_causes(self, causes: 'list[Schema_ErrorCause | tuple[Enum_CauseCo
                 else:
                     code, args = cast('tuple[Enum_CauseCode, dict[str, Any]]', schema)
                     causes_list.append(self._make_sctp_cause(code, None, **args))
-            return causes_list
+            return _last_item_only(causes_list)
 
         for code, cause in causes.items(multi=True):
             causes_list.append(self._make_sctp_cause(code, cause))
-        return causes_list
+        return _last_item_only(causes_list)
 
     def _make_sctp_cause(self, code: 'Enum_CauseCode', cause: 'Optional[Data_ErrorCause]' = None,
                          **kwargs: 'Any') -> 'Schema_ErrorCause':