From 03265271b650f89aa448cf63eda321b77ba575d8 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 8 Oct 2026 15:17:26 -0400 Subject: [PATCH] fix(corekit)!: non-zero padding octets are zeroed on rebuild (#1223) PaddingField now packs the octets it was given, truncated or zero-filled to its width as struct's `s` format does, and zeros when it has none, so a freshly built packet is unchanged. Schema.pack delegates its padding branch to the field. Each site carries the octets through its reader, data model and make(); every maker reads the new field tolerantly, defaulting to zeros: - HOPOPT/IPv6-Opts PadN `pad`, QS report `unused` - IPv6-Frag `reserved_octet`; IPv6-Route SR0/Type 2 `reserved`, RPL `padding` - MH PadN option and Flow-ID PadN sub-option `data` (#1329) - HTTP/2 DATA/HEADERS/PUSH_PROMISE `padding`; OSPF crypto auth `reserved` - HIP: `reserved` on the 12 parameters listed on #1223 - HIP and SCTP trailing alignment padding: kept on the data model as `padding` only when not all zeros, and written back on make Existing tests that build these data models by hand now supply the new fields. --- pcapkit/corekit/fields/strings.py | 24 ++ pcapkit/protocols/application/httpv2.py | 18 ++ pcapkit/protocols/application/ospf.py | 2 + pcapkit/protocols/data/application/httpv2.py | 12 +- pcapkit/protocols/data/application/ospf.py | 4 +- pcapkit/protocols/data/internet/hip.py | 53 ++++- pcapkit/protocols/data/internet/hopopt.py | 10 +- pcapkit/protocols/data/internet/ipv6_frag.py | 4 +- pcapkit/protocols/data/internet/ipv6_opts.py | 10 +- pcapkit/protocols/data/internet/ipv6_route.py | 12 +- pcapkit/protocols/data/internet/mh.py | 12 +- pcapkit/protocols/data/transport/sctp.py | 15 ++ pcapkit/protocols/internet/hip.py | 84 +++++++ pcapkit/protocols/internet/hopopt.py | 11 + pcapkit/protocols/internet/ipv6_frag.py | 6 + pcapkit/protocols/internet/ipv6_opts.py | 11 + pcapkit/protocols/internet/ipv6_route.py | 18 ++ pcapkit/protocols/internet/mh.py | 13 +- .../protocols/schema/application/httpv2.py | 6 +- pcapkit/protocols/schema/application/ospf.py | 2 +- pcapkit/protocols/schema/internet/hip.py | 24 +- pcapkit/protocols/schema/internet/hopopt.py | 4 +- .../protocols/schema/internet/ipv6_frag.py | 2 +- .../protocols/schema/internet/ipv6_opts.py | 4 +- .../protocols/schema/internet/ipv6_route.py | 6 +- pcapkit/protocols/schema/internet/mh.py | 4 +- pcapkit/protocols/schema/schema.py | 5 +- pcapkit/protocols/transport/sctp.py | 54 ++++- tests/protocols/application/test_http_unit.py | 7 +- tests/protocols/application/test_ospf_unit.py | 2 +- tests/protocols/internet/test_hip_unit.py | 24 +- .../internet/test_ipv6_extension_unit.py | 12 +- .../internet/test_mh_padn_contents_unit.py | 72 ++++++ tests/protocols/internet/test_mh_unit.py | 5 +- .../test_padding_kept_as_captured_unit.py | 207 ++++++++++++++++++ 35 files changed, 676 insertions(+), 83 deletions(-) create mode 100644 tests/protocols/internet/test_mh_padn_contents_unit.py create mode 100644 tests/protocols/test_padding_kept_as_captured_unit.py diff --git a/pcapkit/corekit/fields/strings.py b/pcapkit/corekit/fields/strings.py index d8a1deeb20..fcf93254b9 100644 --- a/pcapkit/corekit/fields/strings.py +++ b/pcapkit/corekit/fields/strings.py @@ -348,4 +348,28 @@ class PaddingField(BytesField): callback: Callback function to be called upon :meth:`self.__call__ `. + Padding packs back as the octets it was given, so a schema unpacked from + non-zero padding rebuilds byte for byte. An absent or empty value packs as + zeros -- the default for a freshly built packet -- and a value of another + width is truncated or zero-filled to the field's, as :mod:`struct`'s ``s`` + format does. + """ + + def pack(self, value: 'Optional[bytes]', packet: 'dict[str, Any]') -> 'bytes': # pylint: disable=unused-argument + """Pack field value into :obj:`bytes`. + + Args: + value: Field value, i.e. the padding octets as captured. + packet: Packet data. + + Returns: + Packed field value: ``value`` truncated or zero-filled to the + field's width, or that many zero octets if ``value`` is absent or + empty. + + """ + length = self.length + if isinstance(value, (bytes, bytearray)) and value: + return bytes(value[:length]).ljust(length, b'\x00') + return bytes(length) diff --git a/pcapkit/protocols/application/httpv2.py b/pcapkit/protocols/application/httpv2.py index 4fbfa860bd..3131ed46d5 100644 --- a/pcapkit/protocols/application/httpv2.py +++ b/pcapkit/protocols/application/httpv2.py @@ -611,6 +611,7 @@ def _read_http_data(self, schema: 'Schema_DataFrame', *, reserved=header.stream['reserved'], sid=header.stream['sid'], data=schema.data, + padding=schema.padding if flag.PADDED else b'', ) return data @@ -669,6 +670,7 @@ def _read_http_headers(self, schema: 'Schema_HeadersFrame', *, stream_dependency=schema.stream_dep['sid'] if flag.PRIORITY else 0, weight=(schema.weight + 1) if flag.PRIORITY else 0, fragment=schema.fragment, + padding=schema.padding if flag.PADDED else b'', ) return data @@ -878,6 +880,7 @@ def _read_http_push_promise(self, schema: 'Schema_PushPromiseFrame', *, promised_reserved=schema.stream['reserved'], promised_sid=schema.stream['sid'], fragment=schema.fragment, + padding=schema.padding if flag.PADDED else b'', ) return data @@ -1088,6 +1091,7 @@ def _make_http_data(self, frame: 'Optional[Data_DataFrame]' = None, *, padded: 'bool' = False, pad_len: 'int' = 0, data: 'bytes' = b'', + padding: 'bytes' = b'', **kwargs: 'Any') -> 'tuple[Schema_DataFrame, Flags]': """Make HTTP/2 ``DATA`` frame. @@ -1097,6 +1101,8 @@ def _make_http_data(self, frame: 'Optional[Data_DataFrame]' = None, *, padded: Padded flag, implied by a non-zero ``pad_len``. pad_len: Padding length. data: Frame data. + padding: Padding octets, truncated or zero-filled to ``pad_len`` + octets. **kwargs: Arbitrary keyword arguments. Returns: @@ -1109,6 +1115,7 @@ def _make_http_data(self, frame: 'Optional[Data_DataFrame]' = None, *, pad_len = frame.pad_len data = frame.data + padding = getattr(frame, 'padding', b'') flags = Schema_DataFrame.Flags(0) if end_stream: @@ -1119,6 +1126,7 @@ def _make_http_data(self, frame: 'Optional[Data_DataFrame]' = None, *, return Schema_DataFrame( pad_len=pad_len, data=data, + padding=padding, ), flags def _make_http_headers(self, frame: 'Optional[Data_HeadersFrame]' = None, *, @@ -1130,6 +1138,7 @@ def _make_http_headers(self, frame: 'Optional[Data_HeadersFrame]' = None, *, sid_dep: 'Optional[int]' = None, weight: 'int' = 0, fragment: 'bytes' = b'', + padding: 'bytes' = b'', **kwargs: 'Any') -> 'tuple[Schema_HeadersFrame, Flags]': """Make HTTP/2 ``HEADERS`` frame. @@ -1143,6 +1152,8 @@ def _make_http_headers(self, frame: 'Optional[Data_HeadersFrame]' = None, *, sid_dep: Dependency stream identifier. weight: Priority weight value. fragment: Header block fragment. + padding: Padding octets, truncated or zero-filled to ``pad_len`` + octets. **kwargs: Arbitrary keyword arguments. Returns: @@ -1160,6 +1171,7 @@ def _make_http_headers(self, frame: 'Optional[Data_HeadersFrame]' = None, *, sid_dep = frame.stream_dependency weight = frame.weight fragment = frame.fragment + padding = getattr(frame, 'padding', b'') else: priority = sid_dep is not None sid_dep = sid_dep or 0 @@ -1182,6 +1194,7 @@ def _make_http_headers(self, frame: 'Optional[Data_HeadersFrame]' = None, *, }, weight=weight - 1 if weight else 0, fragment=fragment, + padding=padding, ), flags def _make_http_priority(self, frame: 'Optional[Data_PriorityFrame]' = None, *, @@ -1304,6 +1317,7 @@ def _make_http_push_promise(self, frame: 'Optional[Data_PushPromiseFrame]' = Non promised_reserved: 'int' = 0, promised_sid: 'int' = 0, fragment: 'bytes' = b'', + padding: 'bytes' = b'', **kwargs: 'Any') -> 'tuple[Schema_PushPromiseFrame, Flags]': """Make HTTP/2 ``PUSH_PROMISE`` frame. @@ -1315,6 +1329,8 @@ def _make_http_push_promise(self, frame: 'Optional[Data_PushPromiseFrame]' = Non promised_reserved: Reserved bit of the promised stream identifier. promised_sid: Promised stream identifier. fragment: Header block fragment. + padding: Padding octets, truncated or zero-filled to ``pad_len`` + octets. **kwargs: Arbitrary keyword arguments. Returns: @@ -1328,6 +1344,7 @@ def _make_http_push_promise(self, frame: 'Optional[Data_PushPromiseFrame]' = Non promised_reserved = frame.promised_reserved promised_sid = frame.promised_sid fragment = frame.fragment + padding = getattr(frame, 'padding', b'') flags = Schema_PushPromiseFrame.Flags(0) if end_headers: @@ -1342,6 +1359,7 @@ def _make_http_push_promise(self, frame: 'Optional[Data_PushPromiseFrame]' = Non 'sid': promised_sid, }, fragment=fragment, + padding=padding, ), flags def _make_http_ping(self, frame: 'Optional[Data_PingFrame]' = None, *, diff --git a/pcapkit/protocols/application/ospf.py b/pcapkit/protocols/application/ospf.py index 3344e0ec7f..aeef7c9c70 100644 --- a/pcapkit/protocols/application/ospf.py +++ b/pcapkit/protocols/application/ospf.py @@ -392,6 +392,7 @@ def _read_encrypt_auth(self, schema: 'Schema_CryptographicAuthentication') -> 'D """ auth = Data_CryptographicAuthentication( + reserved=schema.reserved, key_id=schema.key_id, len=schema.len, seq=schema.seq, @@ -422,6 +423,7 @@ def _make_encrypt_auth(self, return Schema_CryptographicAuthentication.unpack(auth_data) # type: ignore[call-arg,misc] if isinstance(auth_data, Data_CryptographicAuthentication): return Schema_CryptographicAuthentication( + reserved=getattr(auth_data, 'reserved', b'\x00\x00'), key_id=auth_data.key_id, len=auth_data.len, seq=auth_data.seq, diff --git a/pcapkit/protocols/data/application/httpv2.py b/pcapkit/protocols/data/application/httpv2.py index 185945abcd..5295d0d95c 100644 --- a/pcapkit/protocols/data/application/httpv2.py +++ b/pcapkit/protocols/data/application/httpv2.py @@ -89,9 +89,11 @@ class DataFrame(HTTP): pad_len: 'int' #: Frame payload. data: 'bytes' + #: Padding octets, as captured (empty unless ``PADDED``). + padding: 'bytes' if TYPE_CHECKING: - def __init__(self, length: 'int', type: 'Frame', flags: 'DataFrameFlags', reserved: 'int', pad_len: 'int', sid: 'int', data: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, length: 'int', type: 'Frame', flags: 'DataFrameFlags', reserved: 'int', pad_len: 'int', sid: 'int', data: 'bytes', padding: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -127,9 +129,11 @@ class HeadersFrame(HTTP): weight: 'int' #: Header block fragment. fragment: 'bytes' + #: Padding octets, as captured (empty unless ``PADDED``). + padding: 'bytes' if TYPE_CHECKING: - def __init__(self, length: 'int', type: 'Frame', flags: 'HeadersFrameFlags', reserved: 'int', pad_len: 'int', sid: 'int', excl_dependency: 'bool', stream_dependency: 'int', weight: 'int', fragment: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, length: 'int', type: 'Frame', flags: 'HeadersFrameFlags', reserved: 'int', pad_len: 'int', sid: 'int', excl_dependency: 'bool', stream_dependency: 'int', weight: 'int', fragment: 'bytes', padding: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -237,9 +241,11 @@ class PushPromiseFrame(HTTP): promised_sid: 'int' #: Header block fragment. fragment: 'bytes' + #: Padding octets, as captured (empty unless ``PADDED``). + padding: 'bytes' if TYPE_CHECKING: - def __init__(self, length: 'int', type: 'Frame', flags: 'Flags', reserved: 'int', pad_len: 'int', sid: 'int', promised_reserved: 'int', promised_sid: 'int', fragment: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, length: 'int', type: 'Frame', flags: 'Flags', reserved: 'int', pad_len: 'int', sid: 'int', promised_reserved: 'int', promised_sid: 'int', fragment: 'bytes', padding: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final diff --git a/pcapkit/protocols/data/application/ospf.py b/pcapkit/protocols/data/application/ospf.py index 1801b3a7b2..c6f9cf5c4b 100644 --- a/pcapkit/protocols/data/application/ospf.py +++ b/pcapkit/protocols/data/application/ospf.py @@ -20,6 +20,8 @@ class CryptographicAuthentication(Data): """Data model for OSPF cryptographic authentication.""" + #: The two reserved octets, as captured. + reserved: 'bytes' #: Key ID. key_id: 'int' #: Authentication data length. @@ -28,7 +30,7 @@ class CryptographicAuthentication(Data): seq: 'int' if TYPE_CHECKING: - def __init__(self, key_id: 'int', len: 'int', seq: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin + def __init__(self, reserved: 'bytes', key_id: 'int', len: 'int', seq: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin @info_final diff --git a/pcapkit/protocols/data/internet/hip.py b/pcapkit/protocols/data/internet/hip.py index 45585dd306..874af5fd70 100644 --- a/pcapkit/protocols/data/internet/hip.py +++ b/pcapkit/protocols/data/internet/hip.py @@ -84,6 +84,11 @@ class Parameter(Data): #: Content length. length: 'int' + if TYPE_CHECKING: + #: Trailing padding octets, as captured; present only when they are + #: not all zeros. + padding: 'bytes' + @info_final class HIP(Protocol): @@ -148,11 +153,13 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', rese class R1CounterParameter(Parameter): """Data model for HIP ``R1_COUNTER`` parameter.""" + #: The four reserved octets, as captured. + reserved: 'bytes' #: R1 counter. counter: 'int' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', counter: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', counter: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -341,11 +348,13 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', ciph class NATTraversalModeParameter(Parameter): """Data model for HIP ``NAT_TRAVERSAL_MODE`` parameter.""" + #: The two reserved octets, as captured. + reserved: 'bytes' #: Mode IDs mode_id: 'tuple[NATTraversal, ...]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', mode_id: 'tuple[NATTraversal, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', mode_id: 'tuple[NATTraversal, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -363,6 +372,8 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', min_ class EncryptedParameter(Parameter): """Data model for HIP ``ENCRYPTED`` parameter.""" + #: The four reserved octets, as captured. + reserved: 'bytes' #: Cipher ID. cipher: 'Cipher' #: Initialization vector. @@ -371,7 +382,7 @@ class EncryptedParameter(Parameter): data: 'bytes' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', cipher: 'Cipher', iv: 'Optional[bytes]', data: 'bytes') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @@ -444,13 +455,15 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', cert class NotificationParameter(Parameter): """Data model for HIP ``NOTIFICATION`` parameter.""" + #: The two reserved octets, as captured. + reserved: 'bytes' #: Notify message type. msg_type: 'NotifyMessage' #: Notification data. msg: 'bytes' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', msg_type: 'NotifyMessage', msg: 'bytes') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', msg_type: 'NotifyMessage', msg: 'bytes') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -533,6 +546,8 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', life class RegFromParameter(Parameter): """Data model for HIP ``REG_FROM`` parameter.""" + #: The reserved octet, as captured. + reserved: 'bytes' #: Port. port: 'int' #: Protocol. @@ -541,7 +556,7 @@ class RegFromParameter(Parameter): address: 'IPv6Address' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', port: 'int', protocol: 'TransType', address: 'IPv6Address') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', port: 'int', protocol: 'TransType', address: 'IPv6Address') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -570,11 +585,13 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', tf_t class ESPTransformParameter(Parameter): """Data model for HIP ``ESP_TRANSFORM`` parameter.""" + #: The two reserved octets, as captured. + reserved: 'bytes' #: ESP transform. suite_id: 'tuple[ESPTransformSuite, ...]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', suite_id: 'tuple[ESPTransformSuite, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', suite_id: 'tuple[ESPTransformSuite, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -603,6 +620,8 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', ack: class PayloadMICParameter(Parameter): """Data model for HIP ``PAYLOAD_MIC`` parameter.""" + #: The three reserved octets, as captured. + reserved: 'bytes' #: Next header next: 'TransType' #: Payload data. @@ -611,7 +630,7 @@ class PayloadMICParameter(Parameter): mic: 'bytes' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', next: 'TransType', payload: 'bytes', mic: 'bytes') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', next: 'TransType', payload: 'bytes', mic: 'bytes') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -656,13 +675,15 @@ def __init__(self, symmetric: 'bool', must_follow: 'bool', reserved: 'int') -> ' class RouteDstParameter(Parameter): """Data model for HIP ``ROUTE_DST`` parameter.""" + #: The two reserved octets, as captured. + reserved: 'bytes' #: Flags. flags: 'Flags' #: Destination address. hit: 'tuple[IPv6Address, ...]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', flags: 'Flags', hit: 'tuple[IPv6Address, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', flags: 'Flags', hit: 'tuple[IPv6Address, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -752,6 +773,8 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', opaq class RelayFromParameter(Parameter): """Data model for HIP ``RELAY_FROM`` parameter.""" + #: The reserved octet, as captured. + reserved: 'bytes' #: Port. port: 'int' #: Protocol. @@ -760,13 +783,15 @@ class RelayFromParameter(Parameter): address: 'IPv6Address' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', port: 'int', protocol: 'TransType', address: 'IPv6Address') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', port: 'int', protocol: 'TransType', address: 'IPv6Address') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final class RelayToParameter(Parameter): """Data model for HIP ``RELAY_TO`` parameter.""" + #: The reserved octet, as captured. + reserved: 'bytes' #: Port. port: 'int' #: Protocol. @@ -775,31 +800,35 @@ class RelayToParameter(Parameter): address: 'IPv6Address' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', port: 'int', protocol: 'TransType', address: 'IPv6Address') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', port: 'int', protocol: 'TransType', address: 'IPv6Address') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final class OverlayTTLParameter(Parameter): """Data model for HIP ``OVERLAY_TTL`` parameter.""" + #: The two reserved octets, as captured. + reserved: 'bytes' #: TTL value. ttl: 'timedelta' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', ttl: 'timedelta') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', ttl: 'timedelta') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final class RouteViaParameter(Parameter): """Data model for HIP ``ROUTE_VIA`` parameter.""" + #: The two reserved octets, as captured. + reserved: 'bytes' #: Flags. flags: 'Flags' #: HIT addresses. hit: 'tuple[IPv6Address, ...]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', flags: 'Flags', hit: 'tuple[IPv6Address, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', flags: 'Flags', hit: 'tuple[IPv6Address, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final diff --git a/pcapkit/protocols/data/internet/hopopt.py b/pcapkit/protocols/data/internet/hopopt.py index 19885361aa..8ad1fcaf14 100644 --- a/pcapkit/protocols/data/internet/hopopt.py +++ b/pcapkit/protocols/data/internet/hopopt.py @@ -80,8 +80,12 @@ def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: ' class PadOption(Option): """Data model for HOPOPT padding options.""" + #: Padding octets, i.e. the ``PadN`` option data as captured (empty for + #: a ``Pad1``). + pad: 'bytes' + if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long + def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', pad: 'bytes') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long @info_final @@ -217,9 +221,11 @@ class QuickStartReportOption(QuickStartOption): nonce: 'int' #: Reserved bits (``R``) after the nonce. reserved: 'int' + #: The "Not Used" octet after the rate, as captured. + unused: 'bytes' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', func: 'QSFunction', rate: 'int', nonce: 'int', reserved: 'int') -> 'None': ... + def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', func: 'QSFunction', rate: 'int', nonce: 'int', reserved: 'int', unused: 'bytes') -> 'None': ... @info_final diff --git a/pcapkit/protocols/data/internet/ipv6_frag.py b/pcapkit/protocols/data/internet/ipv6_frag.py index ceb625a9d8..8004280458 100644 --- a/pcapkit/protocols/data/internet/ipv6_frag.py +++ b/pcapkit/protocols/data/internet/ipv6_frag.py @@ -18,6 +18,8 @@ class IPv6_Frag(Protocol): #: Next header. next: 'TransType' + #: The reserved octet after the next header, as captured. + reserved_octet: 'bytes' #: Fragment offset. offset: 'int' #: Reserved bits (``Res``) between the offset and the M flag, carried @@ -29,4 +31,4 @@ class IPv6_Frag(Protocol): id: 'int' if TYPE_CHECKING: - def __init__(self, next: 'TransType', offset: 'int', reserved: 'int', mf: 'bool', id: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,redefined-builtin,multiple-statements + def __init__(self, next: 'TransType', reserved_octet: 'bytes', offset: 'int', reserved: 'int', mf: 'bool', id: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,redefined-builtin,multiple-statements diff --git a/pcapkit/protocols/data/internet/ipv6_opts.py b/pcapkit/protocols/data/internet/ipv6_opts.py index 007e09dfae..c064f5fb8c 100644 --- a/pcapkit/protocols/data/internet/ipv6_opts.py +++ b/pcapkit/protocols/data/internet/ipv6_opts.py @@ -82,9 +82,13 @@ def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: ' class PadOption(Option): """Data model for IPv6-Opts padding options.""" + #: Padding octets, i.e. the ``PadN`` option data as captured (empty for + #: a ``Pad1``). + pad: 'bytes' + if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', - length: 'int') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long + length: 'int', pad: 'bytes') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,redefined-builtin,multiple-statements,line-too-long @info_final @@ -226,9 +230,11 @@ class QuickStartReportOption(QuickStartOption): nonce: 'int' #: Reserved bits (``R``) after the nonce. reserved: 'int' + #: The "Not Used" octet after the rate, as captured. + unused: 'bytes' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', func: 'QSFunction', rate: 'int', nonce: 'int', reserved: 'int') -> 'None': + def __init__(self, type: 'Enum_Option', action: 'int', change: 'bool', length: 'int', func: 'QSFunction', rate: 'int', nonce: 'int', reserved: 'int', unused: 'bytes') -> 'None': ... diff --git a/pcapkit/protocols/data/internet/ipv6_route.py b/pcapkit/protocols/data/internet/ipv6_route.py index 274a63d6de..a172c5d7ed 100644 --- a/pcapkit/protocols/data/internet/ipv6_route.py +++ b/pcapkit/protocols/data/internet/ipv6_route.py @@ -48,24 +48,28 @@ def __init__(self, next: 'TransType', length: 'int', type: 'Routing', seg_left: class SourceRoute(IPv6_Route): """Data model for IPv6-Route Source Route data type.""" + #: The four reserved octets, as captured. + reserved: 'bytes' #: Source addresses. ip: 'tuple[IPv6Address, ...]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Routing', seg_left: 'int', - ip: 'tuple[IPv6Address, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,super-init-not-called,redefined-builtin,line-too-long + reserved: 'bytes', ip: 'tuple[IPv6Address, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,super-init-not-called,redefined-builtin,line-too-long @info_final class Type2(IPv6_Route): """Data model for IPv6-Route Type 2 data type.""" + #: The four reserved octets, as captured. + reserved: 'bytes' #: Address. ip: 'IPv6Address' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Routing', seg_left: 'int', - ip: 'IPv6Address') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,super-init-not-called,redefined-builtin,line-too-long + reserved: 'bytes', ip: 'IPv6Address') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,super-init-not-called,redefined-builtin,line-too-long @info_final @@ -82,7 +86,9 @@ class RPL(IPv6_Route): reserved: 'int' #: Addresses. ip: 'tuple[IPv6Address | bytes, ...]' + #: The ``Pad`` octets after the addresses, as captured. + padding: 'bytes' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Routing', seg_left: 'int', - cmpr_i: 'int', cmpr_e: 'int', pad: 'int', reserved: 'int', ip: 'tuple[IPv6Address | bytes, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,super-init-not-called,redefined-builtin,line-too-long + cmpr_i: 'int', cmpr_e: 'int', pad: 'int', reserved: 'int', ip: 'tuple[IPv6Address | bytes, ...]', padding: 'bytes') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,super-init-not-called,redefined-builtin,line-too-long diff --git a/pcapkit/protocols/data/internet/mh.py b/pcapkit/protocols/data/internet/mh.py index 7a971715f4..d95460ceda 100644 --- a/pcapkit/protocols/data/internet/mh.py +++ b/pcapkit/protocols/data/internet/mh.py @@ -736,8 +736,12 @@ def __init__(self, type: 'Enum_Option', length: 'int', data: 'bytes') -> 'None': class PadOption(Option): """Data model for Pad option.""" + #: Padding octets, i.e. the ``PadN`` option data as captured (empty for a + #: ``Pad1``). + data: 'bytes' + if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int') -> 'None': ... + def __init__(self, type: 'Enum_Option', length: 'int', data: 'bytes') -> 'None': ... @info_final @@ -1446,8 +1450,12 @@ def __init__(self, type: 'Enum_FlowIDSuboption', length: 'int', class PadFlowIdentificationSuboption(FlowIdentificationSuboption): """Data model for MH Flow Identification padding sub-options.""" + #: Padding octets, i.e. the ``PadN`` sub-option data as captured (empty for + #: a ``Pad``). + data: 'bytes' + if TYPE_CHECKING: - def __init__(self, type: 'Enum_FlowIDSuboption', length: 'int') -> 'None': ... + def __init__(self, type: 'Enum_FlowIDSuboption', length: 'int', data: 'bytes') -> 'None': ... @info_final diff --git a/pcapkit/protocols/data/transport/sctp.py b/pcapkit/protocols/data/transport/sctp.py index 29e22a52b6..88d9dd2d81 100644 --- a/pcapkit/protocols/data/transport/sctp.py +++ b/pcapkit/protocols/data/transport/sctp.py @@ -102,6 +102,11 @@ class ErrorCause(Data): #: Cause length. length: 'int' + if TYPE_CHECKING: + #: Trailing padding octets, as captured; present only when they are + #: not all zeros. + padding: 'bytes' + @info_final class UnknownCause(ErrorCause): @@ -260,6 +265,11 @@ class Parameter(Data): #: Parameter length. length: 'int' + if TYPE_CHECKING: + #: Trailing padding octets, as captured; present only when they are + #: not all zeros. + padding: 'bytes' + @info_final class UnknownParameter(Parameter): @@ -368,6 +378,11 @@ class Chunk(Data): #: Chunk length, excluding any trailing padding. length: 'int' + if TYPE_CHECKING: + #: Trailing padding octets, as captured; present only when they are + #: not all zeros. + padding: 'bytes' + @info_final class UnknownChunk(Chunk): diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index b810dd35b6..637ffc2a68 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -820,6 +820,14 @@ def _read_hip_param(self, length: 'int', *, version: 'int') -> 'Parameter': # p meth = name[0] data = meth(schema, version=version, options=options) + # NOTE: The trailing padding (:rfc:`7401#section-5.2.1`) is kept on + # the data model only when it is not all zeros -- the form a fresh + # build writes -- so that a rebuild reproduces it (:issue:`1223`). + # :meth:`_make_hip_param` writes it back. + padding = schema.__dict__.get('padding') + if isinstance(padding, bytes) and any(padding): + data.__update__([('padding', padding)]) + # record parameter data options.add(dscp, data) counter += len(schema) @@ -951,6 +959,7 @@ def _read_param_r1_counter(self, schema: 'Schema_R1CounterParameter', *, version type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, counter=schema.counter, ) return r1_counter @@ -1515,6 +1524,7 @@ def _read_param_nat_traversal_mode(self, schema: 'Schema_NATTraversalModeParamet type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, mode_id=tuple(schema.modes), ) return nat_traversal_mode @@ -1596,6 +1606,7 @@ def _read_param_encrypted(self, schema: 'Schema_EncryptedParameter', *, version: type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, cipher=schema.cipher, iv=getattr(schema, 'iv', None), data=schema.data, @@ -1783,6 +1794,7 @@ def _read_param_notification(self, schema: 'Schema_NotificationParameter', *, ve type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, msg_type=schema.msg_type, msg=schema.msg, ) @@ -2016,6 +2028,7 @@ def _read_param_reg_from(self, schema: 'Schema_RegFromParameter', *, version: 'i type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, port=schema.port, protocol=schema.protocol, address=schema.address, @@ -2138,6 +2151,7 @@ def _read_param_esp_transform(self, schema: 'Schema_ESPTransformParameter', *, v type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, suite_id=tuple(schema.suites), ) return esp_transform @@ -2257,6 +2271,7 @@ def _read_param_payload_mic(self, schema: 'Schema_PayloadMICParameter', *, versi type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, next=schema.next, payload=schema.payload, mic=schema.mic, @@ -2381,6 +2396,7 @@ def _read_param_route_dst(self, schema: 'Schema_RouteDstParameter', *, version: type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, flags=Data_Flags( symmetric=bool(schema.flags['symmetric']), must_follow=bool(schema.flags['must_follow']), @@ -2689,6 +2705,7 @@ def _read_param_relay_from(self, schema: 'Schema_RelayFromParameter', *, version type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, port=schema.port, protocol=schema.protocol, address=address, # type: ignore[arg-type] @@ -2738,6 +2755,7 @@ def _read_param_relay_to(self, schema: 'Schema_RelayToParameter', *, version: 'i type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, port=schema.port, protocol=schema.protocol, address=address, # type: ignore[arg-type] @@ -2779,6 +2797,7 @@ def _read_param_overlay_ttl(self, schema: 'Schema_OverlayTTLParameter', *, versi type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, ttl=datetime.timedelta(seconds=schema.ttl), ) return overlay_ttl @@ -2831,6 +2850,7 @@ def _read_param_route_via(self, schema: 'Schema_RouteViaParameter', *, version: type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), + reserved=schema.reserved, flags=Data_Flags( symmetric=bool(schema.flags['symmetric']), must_follow=bool(schema.flags['must_follow']), @@ -3058,6 +3078,10 @@ def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parame meth = name[1] data = meth(code, param, version=version) + # NOTE: The trailing padding :meth:`_read_hip_param` kept, if any. + padding = getattr(param, 'padding', None) + if isinstance(padding, bytes) and 'padding' in data.__fields__: + data.padding = padding data_packed = data.pack() parameters_list.append(data) @@ -3131,6 +3155,7 @@ def _make_param_esp_info(self, code: 'Enum_Parameter', param: 'Optional[Data_ESP def _make_param_r1_counter(self, code: 'Enum_Parameter', param: 'Optional[Data_R1CounterParameter]' = None, *, # pylint: disable=unused-argument version: 'int', counter: 'int' = 0, + reserved: 'bytes' = b'\x00\x00\x00\x00', **kwargs: 'Any') -> 'Schema_R1CounterParameter': """Make HIP ``R1_COUNTER`` parameter. @@ -3139,6 +3164,8 @@ def _make_param_r1_counter(self, code: 'Enum_Parameter', param: 'Optional[Data_R param: parameter data version: HIP protocol version counter: R1 generation counter + reserved: the four reserved octets, truncated or + zero-filled to 4 octets **kwargs: arbitrary keyword arguments Returns: @@ -3149,9 +3176,11 @@ def _make_param_r1_counter(self, code: 'Enum_Parameter', param: 'Optional[Data_R raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid parameter') if param is not None: + reserved = getattr(param, 'reserved', b'\x00\x00\x00\x00') counter = param.counter return Schema_R1CounterParameter( + reserved=reserved, type=code, len=12, counter=counter, @@ -3773,6 +3802,7 @@ def _make_param_nat_traversal_mode(self, code: 'Enum_Parameter', param: 'Optiona mode_default: 'Optional[int]' = None, mode_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long mode_reversed: 'bool' = False, + reserved: 'bytes' = b'\x00\x00', **kwargs: 'Any') -> 'Schema_NATTraversalModeParameter': """Make HIP ``NAT_TRAVERSAL_MODE`` parameter. @@ -3784,6 +3814,8 @@ def _make_param_nat_traversal_mode(self, code: 'Enum_Parameter', param: 'Optiona mode_default: default mode ID mode_namespace: mode ID namespace mode_reversed: reverse mode ID namespace + reserved: the two reserved octets, truncated or + zero-filled to 2 octets **kwargs: arbitrary keyword arguments Returns: @@ -3791,6 +3823,7 @@ def _make_param_nat_traversal_mode(self, code: 'Enum_Parameter', param: 'Optiona """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00\x00') mode_id = cast('list[Enum_NATTraversal]', param.mode_id) else: if modes is None: @@ -3802,6 +3835,7 @@ def _make_param_nat_traversal_mode(self, code: 'Enum_Parameter', param: 'Optiona reversed=mode_reversed, pack=False)) return Schema_NATTraversalModeParameter( + reserved=reserved, type=code, len=2 + 2 * len(mode_id), modes=mode_id, @@ -3841,6 +3875,7 @@ def _make_param_encrypted(self, code: 'Enum_Parameter', param: 'Optional[Data_En cipher_reversed: 'bool' = False, iv: 'Optional[bytes]' = None, data: 'bytes' = b'', + reserved: 'bytes' = b'\x00\x00\x00\x00', **kwargs: 'Any') -> 'Schema_EncryptedParameter': """Make HIP ``ENCRYPTED`` parameter. @@ -3854,12 +3889,15 @@ def _make_param_encrypted(self, code: 'Enum_Parameter', param: 'Optional[Data_En cipher_reversed: reverse cipher ID namespace iv: initialization vector (optional depending on cipher ID) data: encrypted data + reserved: the four reserved octets, truncated or + zero-filled to 4 octets Returns: HIP parameter schema. """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00\x00\x00\x00') cipher_id = param.cipher iv = param.iv data = param.data @@ -3874,6 +3912,7 @@ def _make_param_encrypted(self, code: 'Enum_Parameter', param: 'Optional[Data_En raise ProtocolError(f'HIPv{version}: [ParamNo {code}] IV length must be 16 bytes for AES cipher') schema = Schema_EncryptedParameter( + reserved=reserved, type=code, len=4 + len(iv or b'') + len(data), iv=iv, @@ -4124,6 +4163,7 @@ def _make_param_notification(self, code: 'Enum_Parameter', param: 'Optional[Data msg_type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long msg_type_reversed: 'bool' = False, msg: 'bytes' = b'', + reserved: 'bytes' = b'\x00\x00', **kwargs: 'Any') -> 'Schema_NotificationParameter': """Make HIP ``NOTIFICATION`` parameter. @@ -4136,6 +4176,8 @@ def _make_param_notification(self, code: 'Enum_Parameter', param: 'Optional[Data msg_type_namespace: notification message type namespace msg_type_reversed: reverse notification message type namespace msg: notification message + reserved: the two reserved octets, truncated or + zero-filled to 2 octets **kwargs: arbitrary keyword arguments Returns: @@ -4143,6 +4185,7 @@ def _make_param_notification(self, code: 'Enum_Parameter', param: 'Optional[Data """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00\x00') type = param.msg_type msg = param.msg else: @@ -4150,6 +4193,7 @@ def _make_param_notification(self, code: 'Enum_Parameter', param: 'Optional[Data reversed=msg_type_reversed, pack=False) return Schema_NotificationParameter( + reserved=reserved, type=code, len=4 + len(msg), msg_type=type, @@ -4377,6 +4421,7 @@ def _make_param_reg_from(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg protocol_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long protocol_reversed: 'bool' = False, address: 'IPv6Address | str | int | bytes' = '::', + reserved: 'bytes' = b'\x00', **kwargs: 'Any') -> 'Schema_RegFromParameter': """Make HIP ``REG_FROM`` parameter. @@ -4390,6 +4435,8 @@ def _make_param_reg_from(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg protocol_namespace: transport protocol namespace protocol_reversed: reverse transport protocol namespace address: IPv6 address + reserved: the reserved octet, truncated or + zero-filled to 1 octet **kwargs: arbitrary keyword arguments Returns: @@ -4397,6 +4444,7 @@ def _make_param_reg_from(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00') port = param.port proto = param.protocol address = param.address @@ -4405,6 +4453,7 @@ def _make_param_reg_from(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg reversed=protocol_reversed, pack=False) return Schema_RegFromParameter( + reserved=reserved, type=code, len=20, port=port, @@ -4490,6 +4539,7 @@ def _make_param_esp_transform(self, code: 'Enum_Parameter', param: 'Optional[Dat suite_default: 'Optional[int]' = None, suite_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long suite_reversed: 'bool' = False, + reserved: 'bytes' = b'\x00\x00', **kwargs: 'Any') -> 'Schema_ESPTransformParameter': """Make HIP ``ESP_TRANSFORM`` parameter. @@ -4501,6 +4551,8 @@ def _make_param_esp_transform(self, code: 'Enum_Parameter', param: 'Optional[Dat suite_default: default ESP transform suite suite_namespace: ESP transform suite namespace suite_reversed: reverse ESP transform suite namespace + reserved: the two reserved octets, truncated or + zero-filled to 2 octets **kwargs: arbitrary keyword arguments Returns: @@ -4508,6 +4560,7 @@ def _make_param_esp_transform(self, code: 'Enum_Parameter', param: 'Optional[Dat """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00\x00') suite_id = cast('list[Enum_ESPTransformSuite]', param.suite_id) else: if suites is None: @@ -4519,6 +4572,7 @@ def _make_param_esp_transform(self, code: 'Enum_Parameter', param: 'Optional[Dat reversed=suite_reversed, pack=False)) return Schema_ESPTransformParameter( + reserved=reserved, type=code, len=2 + 2 * len(suite_id), suites=suite_id, @@ -4585,6 +4639,7 @@ def _make_param_payload_mic(self, code: 'Enum_Parameter', param: 'Optional[Data_ next_reversed: 'bool' = False, payload: 'bytes' = b'', mic: 'bytes' = b'', + reserved: 'bytes' = b'\x00\x00\x00', **kwargs: 'Any') -> 'Schema_PayloadMICParameter': """Make HIP ``PAYLOAD_MIC`` parameter. @@ -4598,6 +4653,8 @@ def _make_param_payload_mic(self, code: 'Enum_Parameter', param: 'Optional[Data_ next_reversed: reverse next protocol namespace payload: payload data mic: message integrity code + reserved: the three reserved octets, truncated or + zero-filled to 3 octets **kwargs: arbitrary keyword arguments Returns: @@ -4605,6 +4662,7 @@ def _make_param_payload_mic(self, code: 'Enum_Parameter', param: 'Optional[Data_ """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00\x00\x00') protocol = param.next payload = param.payload mic = param.mic @@ -4613,6 +4671,7 @@ def _make_param_payload_mic(self, code: 'Enum_Parameter', param: 'Optional[Data_ reversed=next_reversed, pack=False) return Schema_PayloadMICParameter( + reserved=reserved, type=code, len=8 + len(mic), next=protocol, @@ -4678,6 +4737,7 @@ def _make_param_route_dst(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro must_follow: 'bool' = False, flags_reserved: 'int' = 0, hit: 'Optional[list[bytes | str | int | IPv6Address]]' = None, + reserved: 'bytes' = b'\x00\x00', **kwargs: 'Any') -> 'Schema_RouteDstParameter': """Make HIP ``ROUTE_DST`` parameter. @@ -4689,12 +4749,15 @@ def _make_param_route_dst(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro must_follow: must-follow flag flags_reserved: the 14 reserved low-order bits of ``Flags`` hit: list of HITs + reserved: the two reserved octets, truncated or + zero-filled to 2 octets Returns: HIP parameter schema. """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00\x00') symmetric = param.flags.symmetric must_follow = param.flags.must_follow flags_reserved = param.flags.reserved @@ -4703,6 +4766,7 @@ def _make_param_route_dst(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro hit_list = hit if hit is not None else [] return Schema_RouteDstParameter( + reserved=reserved, type=code, len=4 + 16 * len(hit_list), flags={ @@ -4945,6 +5009,7 @@ def _make_param_relay_from(self, code: 'Enum_Parameter', param: 'Optional[Data_R protocol_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long protocol_reversed: 'bool' = False, address: 'IPv6Address | str | int | bytes' = '::', + reserved: 'bytes' = b'\x00', **kwargs: 'Any') -> 'Schema_RelayFromParameter': """Make HIP ``RELAY_FROM`` parameter. @@ -4958,12 +5023,15 @@ def _make_param_relay_from(self, code: 'Enum_Parameter', param: 'Optional[Data_R protocol_namespace: transport protocol namespace protocol_reversed: reverse transport protocol namespace address: relay address + reserved: the reserved octet, truncated or + zero-filled to 1 octet Returns: HIP parameter schema. """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00') port = param.port proto = param.protocol address = param.address @@ -4972,6 +5040,7 @@ def _make_param_relay_from(self, code: 'Enum_Parameter', param: 'Optional[Data_R reversed=protocol_reversed, pack=False) return Schema_RelayFromParameter( + reserved=reserved, type=code, len=20, port=port, @@ -4987,6 +5056,7 @@ def _make_param_relay_to(self, code: 'Enum_Parameter', param: 'Optional[Data_Rel protocol_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long protocol_reversed: 'bool' = False, address: 'IPv6Address | str | int | bytes' = '::', + reserved: 'bytes' = b'\x00', **kwargs: 'Any') -> 'Schema_RelayToParameter': """Make HIP ``RELAY_TO`` parameter. @@ -5000,6 +5070,8 @@ def _make_param_relay_to(self, code: 'Enum_Parameter', param: 'Optional[Data_Rel protocol_namespace: transport protocol namespace protocol_reversed: reverse transport protocol namespace address: relay address + reserved: the reserved octet, truncated or + zero-filled to 1 octet **kwargs: arbitrary keyword arguments Returns: @@ -5007,6 +5079,7 @@ def _make_param_relay_to(self, code: 'Enum_Parameter', param: 'Optional[Data_Rel """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00') port = param.port proto = param.protocol address = param.address @@ -5015,6 +5088,7 @@ def _make_param_relay_to(self, code: 'Enum_Parameter', param: 'Optional[Data_Rel reversed=protocol_reversed, pack=False) return Schema_RelayToParameter( + reserved=reserved, type=code, len=20, port=port, @@ -5025,6 +5099,7 @@ def _make_param_relay_to(self, code: 'Enum_Parameter', param: 'Optional[Data_Rel def _make_param_overlay_ttl(self, code: 'Enum_Parameter', param: 'Optional[Data_OverlayTTLParameter]' = None, *, # pylint: disable=unused-argument version: 'int', ttl: 'int | timedelta' = 0, + reserved: 'bytes' = b'\x00\x00', **kwargs: 'Any') -> 'Schema_OverlayTTLParameter': """Make HIP ``OVERLAY_TTL`` parameter. @@ -5033,6 +5108,8 @@ def _make_param_overlay_ttl(self, code: 'Enum_Parameter', param: 'Optional[Data_ param: parameter data version: HIP protocol version ttl: overlay time-to-live (TTL) value + reserved: the two reserved octets, truncated or + zero-filled to 2 octets **kwargs: arbitrary keyword arguments Returns: @@ -5040,11 +5117,13 @@ def _make_param_overlay_ttl(self, code: 'Enum_Parameter', param: 'Optional[Data_ """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00\x00') ttl_val = math.floor(param.ttl.total_seconds()) else: ttl_val = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds()) return Schema_OverlayTTLParameter( + reserved=reserved, type=code, len=4, ttl=ttl_val, @@ -5056,6 +5135,7 @@ def _make_param_route_via(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro must_follow: 'bool' = False, flags_reserved: 'int' = 0, hit: 'Optional[list[IPv6Address | bytes | str | int]]' = None, + reserved: 'bytes' = b'\x00\x00', **kwargs: 'Any') -> 'Schema_RouteViaParameter': """Make HIP ``ROUTE_VIA`` parameter. @@ -5067,12 +5147,15 @@ def _make_param_route_via(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro must_follow: must-follow flag flags_reserved: the 14 reserved low-order bits of ``Flags`` hit: list of HITs + reserved: the two reserved octets, truncated or + zero-filled to 2 octets Returns: HIP parameter schema. """ if param is not None: + reserved = getattr(param, 'reserved', b'\x00\x00') symmetric = param.flags.symmetric must_follow = param.flags.must_follow flags_reserved = param.flags.reserved @@ -5081,6 +5164,7 @@ def _make_param_route_via(self, code: 'Enum_Parameter', param: 'Optional[Data_Ro hit_list = hit if hit is not None else [] return Schema_RouteViaParameter( + reserved=reserved, type=code, len=4 + 16 * len(hit_list), flags={ diff --git a/pcapkit/protocols/internet/hopopt.py b/pcapkit/protocols/internet/hopopt.py index 48b14ecb0e..27e9c86633 100644 --- a/pcapkit/protocols/internet/hopopt.py +++ b/pcapkit/protocols/internet/hopopt.py @@ -652,6 +652,7 @@ def _read_opt_pad(self, schema: 'Schema_PadOption', *, options: 'Option') -> 'Da action=Enum_OptionAction.get(schema.type >> 6), change=bool(schema.type & 0b00100000), length=_size, + pad=schema.pad, ) return opt @@ -991,6 +992,7 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') rate=40000 * (2 ** rate) / 1000 if rate > 0 else 0, nonce=schema_rep.nonce['nonce'], reserved=schema_rep.nonce['reserved'], + unused=schema_rep.reserved, ) else: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] unknown QS function: {func}') @@ -1476,6 +1478,7 @@ def _make_opt_none(self, code: 'Enum_Option', opt: 'Optional[Data_UnassignedOpti def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = None, *, length: 'int' = 0, + pad: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_PadOption': """Make HOPOPT pad option. @@ -1484,6 +1487,8 @@ def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = N opt: option data length: value of the ``Opt Data Len`` field, i.e. the number of padding octets *after* the two octets of the option header + pad: the ``PadN`` option data, truncated or zero-filled to + ``length`` octets **kwargs: arbitrary keyword arguments Returns: @@ -1503,6 +1508,7 @@ def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = N if opt is not None: code = opt.type length = 0 if opt.type == Enum_Option.Pad1 else opt.length - 2 + pad = getattr(opt, 'pad', b'') if code == Enum_Option.Pad1 and length != 0: #raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') @@ -1512,6 +1518,7 @@ def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = N return Schema_PadOption( type=code, len=length, + pad=pad, ) def _make_opt_tun(self, code: 'Enum_Option', opt: 'Optional[Data_TunnelEncapsulationLimitOption]' = None, *, @@ -1843,6 +1850,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption ttl: 'timedelta | int' = 0, nonce: 'int' = 0, reserved: 'int' = 0, + unused: 'bytes' = b'\x00', **kwargs: 'Any') -> 'Schema_QuickStartOption | Schema_UnassignedOption': """Make HOPOPT QS option. @@ -1857,6 +1865,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption ttl: time to live (in seconds) nonce: nonce value reserved: reserved bits (``R``) after the nonce + unused: the "Not Used" octet of a report of approved rate **kwargs: arbitrary keyword arguments Returns: @@ -1872,6 +1881,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption ttl = getattr(opt, 'ttl', 0) nonce = getattr(opt, 'nonce', 0) reserved = getattr(opt, 'reserved', 0) + unused = getattr(opt, 'unused', b'\x00') else: func_enum = self._make_index(func, func_default, namespace=func_namespace, # type: ignore[assignment] reversed=func_reversed, pack=False) @@ -1901,6 +1911,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption 'func': func_enum, 'rate': rate_val, }, + reserved=unused, nonce={ 'nonce': nonce, 'reserved': reserved, diff --git a/pcapkit/protocols/internet/ipv6_frag.py b/pcapkit/protocols/internet/ipv6_frag.py index 328d6d8c39..6f5df5aa30 100644 --- a/pcapkit/protocols/internet/ipv6_frag.py +++ b/pcapkit/protocols/internet/ipv6_frag.py @@ -144,6 +144,7 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, # # exactly as :meth:`pcapkit.protocols.internet.ipv4.IPv4.read` does. ipv6_frag = Data_IPv6_Frag( next=schema.next, + reserved_octet=schema.reserved, offset=int(schema.flags['offset']) * 8, reserved=schema.flags['reserved'], mf=bool(schema.flags['mf']), @@ -159,6 +160,7 @@ def make(self, next_default: 'Optional[int]' = None, next_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long next_reversed: 'bool' = False, + reserved_octet: 'bytes' = b'\x00', offset: 'int' = 0, reserved: 'int' = 0, mf: 'bool' = False, @@ -172,6 +174,8 @@ def make(self, next_default: Default value of next header. next_namespace: Namespace of next header. next_reversed: If the namespace of next header is reversed. + reserved_octet: The reserved octet after the next header, truncated or + zero-filled to one octet. offset: Fragment offset, in on-wire 8-octet units (:rfc:`8200#section-4.5`). Note that :attr:`Data_IPv6_Frag.offset ` is in @@ -191,6 +195,7 @@ def make(self, return Schema_IPv6_Frag( next=next_val, # type: ignore[arg-type] + reserved=reserved_octet, flags={ 'offset': offset, 'reserved': reserved, @@ -267,6 +272,7 @@ def _make_data(cls, data: 'Data_IPv6_Frag') -> 'dict[str, Any]': # type: ignore # keep the data-to-schema round trip exact. return { 'next': data.next, + 'reserved_octet': getattr(data, 'reserved_octet', b'\x00'), 'offset': data.offset // 8, 'reserved': data.reserved, 'mf': data.mf, diff --git a/pcapkit/protocols/internet/ipv6_opts.py b/pcapkit/protocols/internet/ipv6_opts.py index 4d349e7095..5376e9ed0b 100644 --- a/pcapkit/protocols/internet/ipv6_opts.py +++ b/pcapkit/protocols/internet/ipv6_opts.py @@ -641,6 +641,7 @@ def _read_opt_pad(self, schema: 'Schema_PadOption', *, options: 'Option') -> 'Da action=Enum_OptionAction.get(schema.type >> 6), change=bool(schema.type & 0b00100000), length=_size, + pad=schema.pad, ) return opt @@ -980,6 +981,7 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option') rate=40000 * (2 ** rate) / 1000 if rate > 0 else 0, nonce=schema_rep.nonce['nonce'], reserved=schema_rep.nonce['reserved'], + unused=schema_rep.reserved, ) else: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] unknown QS function: {func}') @@ -1465,6 +1467,7 @@ def _make_opt_none(self, code: 'Enum_Option', opt: 'Optional[Data_UnassignedOpti def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = None, *, length: 'int' = 0, + pad: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_PadOption': """Make IPv6-Opts pad option. @@ -1473,6 +1476,8 @@ def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = N opt: option data length: value of the ``Opt Data Len`` field, i.e. the number of padding octets *after* the two octets of the option header + pad: the ``PadN`` option data, truncated or zero-filled to + ``length`` octets **kwargs: arbitrary keyword arguments Returns: @@ -1492,6 +1497,7 @@ def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = N if opt is not None: code = opt.type length = 0 if opt.type == Enum_Option.Pad1 else opt.length - 2 + pad = getattr(opt, 'pad', b'') if code == Enum_Option.Pad1 and length != 0: #raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') @@ -1501,6 +1507,7 @@ def _make_opt_pad(self, code: 'Enum_Option', opt: 'Optional[Data_PadOption]' = N return Schema_PadOption( type=code, len=length, + pad=pad, ) def _make_opt_tun(self, code: 'Enum_Option', opt: 'Optional[Data_TunnelEncapsulationLimitOption]' = None, *, @@ -1832,6 +1839,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption ttl: 'timedelta | int' = 0, nonce: 'int' = 0, reserved: 'int' = 0, + unused: 'bytes' = b'\x00', **kwargs: 'Any') -> 'Schema_QuickStartOption | Schema_UnassignedOption': """Make IPv6-Opts QS option. @@ -1846,6 +1854,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption ttl: time to live (in seconds) nonce: nonce value reserved: reserved bits (``R``) after the nonce + unused: the "Not Used" octet of a report of approved rate **kwargs: arbitrary keyword arguments Returns: @@ -1861,6 +1870,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption ttl = getattr(opt, 'ttl', 0) nonce = getattr(opt, 'nonce', 0) reserved = getattr(opt, 'reserved', 0) + unused = getattr(opt, 'unused', b'\x00') else: func_enum = self._make_index(func, func_default, namespace=func_namespace, # type: ignore[assignment] reversed=func_reversed, pack=False) @@ -1890,6 +1900,7 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption 'func': func_enum, 'rate': rate_val, }, + reserved=unused, nonce={ 'nonce': nonce, 'reserved': reserved, diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index e2235bd5cb..adb4179ef1 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/pcapkit/protocols/internet/ipv6_route.py @@ -549,6 +549,7 @@ def _read_data_type_src(self, schema: 'Schema_SourceRoute', *, header: 'Schema_I length=ipv6_route_header_length(header.length), type=header.type, seg_left=header.seg_left, + reserved=schema.reserved, ip=tuple(schema.ip), ) return ipv6_route @@ -597,6 +598,7 @@ def _read_data_type_2(self, schema: 'Schema_Type2', *, header: 'Schema_IPv6_Rout length=ipv6_route_header_length(header.length), type=header.type, seg_left=header.seg_left, + reserved=schema.reserved, ip=schema.ip, ) return ipv6_route @@ -666,6 +668,7 @@ def _read_data_type_rpl(self, schema: 'Schema_RPL', *, header: 'Schema_IPv6_Rout pad=pad_len, reserved=schema.pad['reserved'], ip=tuple(schema.ip), + padding=schema.padding, ) return ipv6_route @@ -700,6 +703,7 @@ def _make_data_type_none(self, type: 'Enum_Routing', route: 'Optional[Data_Unkno def _make_data_type_src(self, type: 'Enum_Routing', route: 'Optional[Data_SourceRoute]' = None, *, dst: 'Optional[IPv6Address]' = None, ip: 'Optional[list[IPv6Address | str | bytes | int]]' = None, + reserved: 'bytes' = b'\x00\x00\x00\x00', **kwargs: 'Any') -> 'Schema_SourceRoute': """Make IPv6-Route Source Route data. @@ -708,6 +712,8 @@ def _make_data_type_src(self, type: 'Enum_Routing', route: 'Optional[Data_Source route: route data dst: destination IPv6 address ip: list of IPv6 addresses + reserved: the four reserved octets, truncated or zero-filled to + four octets **kwargs: arbitrary keyword arguments Returns: @@ -716,16 +722,19 @@ def _make_data_type_src(self, type: 'Enum_Routing', route: 'Optional[Data_Source """ if route is not None: ip = cast('list[IPv6Address | str | bytes | int]', route.ip) + reserved = getattr(route, 'reserved', b'\x00\x00\x00\x00') else: ip = [] if ip is None else ip return Schema_SourceRoute( + reserved=reserved, ip=ip, ) def _make_data_type_2(self, type: 'Enum_Routing', route: 'Optional[Data_Type2]' = None, *, dst: 'Optional[IPv6Address]' = None, ip: 'IPv6Address | str | bytes | int' = '::', + reserved: 'bytes' = b'\x00\x00\x00\x00', **kwargs: 'Any') -> 'Schema_Type2': """Make IPv6-Route Type 2 data. @@ -734,6 +743,8 @@ def _make_data_type_2(self, type: 'Enum_Routing', route: 'Optional[Data_Type2]' route: route data dst: destination IPv6 address ip: home address + reserved: the four reserved octets, truncated or zero-filled to + four octets **kwargs: arbitrary keyword arguments Returns: @@ -742,8 +753,10 @@ def _make_data_type_2(self, type: 'Enum_Routing', route: 'Optional[Data_Type2]' """ if route is not None: ip = cast('IPv6Address | str | bytes | int', route.ip) + reserved = getattr(route, 'reserved', b'\x00\x00\x00\x00') return Schema_Type2( + reserved=reserved, ip=ip, ) @@ -751,6 +764,7 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' dst: 'Optional[IPv6Address]' = None, ip: 'Optional[list[IPv6Address | str | bytes | int]]' = None, reserved: 'int' = 0, + padding: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_RPL': """Make IPv6-Route RPL Source data. @@ -760,6 +774,8 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' dst: destination IPv6 address ip: list of IPv6 addresses reserved: reserved bits after ``Pad`` + padding: the ``Pad`` octets after the addresses, truncated or + zero-filled to ``Pad`` octets **kwargs: arbitrary keyword arguments Returns: @@ -775,6 +791,7 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' cmpr_e = route.cmpr_e pad = route.pad reserved = route.reserved + padding = getattr(route, 'padding', b'') ip_val = [ addr if isinstance(addr, bytes) else addr.packed for addr in route.ip ] @@ -844,4 +861,5 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' 'reserved': reserved, }, addresses=ip_val, + padding=padding, ) diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 5bf25b0b9a..db8750d46a 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -3013,6 +3013,7 @@ def _read_opt_pad(self, schema: 'Schema_PadOption', *, data = Data_PadOption( type=schema.type, length=size, + data=schema.data, ) return data @@ -4852,6 +4853,7 @@ def _read_fid_suboptions( data = Data_PadFlowIdentificationSuboption( type=code, length=size, + data=pad.data, ) # type: Data_FlowIdentificationSuboption elif code == Enum_FlowIDSuboption.BID_Reference: bid_ref = cast('Schema_BIDReferenceSuboption', schema) @@ -7824,6 +7826,7 @@ def _make_opt_none(self, type: 'Enum_Option', option: 'Optional[Data_UnassignedO def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' = None, *, length: 'int' = 0, + data: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_PadOption': """Make MH pad option. @@ -7832,6 +7835,8 @@ def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' option: Option data model. length: Value of the ``Option Length`` field, i.e. the number of padding octets *after* the two octets of the option header. + data: The ``PadN`` option data, truncated or zero-filled to + ``length`` octets. **kwargs: Arbitrary keyword arguments. Returns: @@ -7852,6 +7857,7 @@ def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' """ if option is not None: length = 0 if option.type == Enum_Option.Pad1 else option.length - 2 + data = getattr(option, 'data', b'') if type == Enum_Option.Pad1 and length != 0: # raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid format') @@ -7861,6 +7867,7 @@ def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' return Schema_PadOption( type=type, length=length, + data=data, ) def _make_opt_bra(self, type: 'Enum_Option', option: 'Optional[Data_BindingRefreshAdviceOption]' = None, *, @@ -9569,7 +9576,11 @@ def _make_fid_suboption(self, code: 'Enum_FlowIDSuboption', pad_len = option.length - 2 else: pad_len = cast('int', kwargs.get('length', 0)) - return Schema_PadFlowIdentificationSuboption(type=code, length=pad_len) + if option is not None: + pad_data = getattr(option, 'data', b'') + else: + pad_data = cast('bytes', kwargs.get('data', b'')) + return Schema_PadFlowIdentificationSuboption(type=code, length=pad_len, data=pad_data) if code == Enum_FlowIDSuboption.BID_Reference: if option is not None: diff --git a/pcapkit/protocols/schema/application/httpv2.py b/pcapkit/protocols/schema/application/httpv2.py index 18d1f04ad1..00b5e0ca7b 100644 --- a/pcapkit/protocols/schema/application/httpv2.py +++ b/pcapkit/protocols/schema/application/httpv2.py @@ -242,7 +242,7 @@ class Flags(FrameType.Flags): ) if TYPE_CHECKING: - def __init__(self, pad_len: 'Optional[int]', data: 'bytes') -> 'None': ... + def __init__(self, pad_len: 'Optional[int]', data: 'bytes', padding: 'Optional[bytes]') -> 'None': ... @schema_final @@ -291,7 +291,7 @@ class Flags(FrameType.Flags): ) if TYPE_CHECKING: - def __init__(self, pad_len: 'Optional[int]', stream_dep: 'Optional[StreamDependency]', weight: 'Optional[int]', fragment: 'bytes') -> 'None': ... + def __init__(self, pad_len: 'Optional[int]', stream_dep: 'Optional[StreamDependency]', weight: 'Optional[int]', fragment: 'bytes', padding: 'Optional[bytes]') -> 'None': ... @schema_final @@ -389,7 +389,7 @@ class Flags(FrameType.Flags): ) if TYPE_CHECKING: - def __init__(self, pad_len: 'Optional[int]', stream: 'StreamID', fragment: 'bytes') -> 'None': ... + def __init__(self, pad_len: 'Optional[int]', stream: 'StreamID', fragment: 'bytes', padding: 'Optional[bytes]') -> 'None': ... @schema_final diff --git a/pcapkit/protocols/schema/application/ospf.py b/pcapkit/protocols/schema/application/ospf.py index fd6f0ff03e..0551f96d7c 100644 --- a/pcapkit/protocols/schema/application/ospf.py +++ b/pcapkit/protocols/schema/application/ospf.py @@ -53,7 +53,7 @@ class CryptographicAuthentication(Schema): seq: 'int' = UInt32Field() if TYPE_CHECKING: - def __init__(self, key_id: 'int', len: 'int', seq: 'int') -> 'None': ... + def __init__(self, reserved: 'bytes', key_id: 'int', len: 'int', seq: 'int') -> 'None': ... @schema_final diff --git a/pcapkit/protocols/schema/internet/hip.py b/pcapkit/protocols/schema/internet/hip.py index c1755e3cb4..7b1e6945ba 100644 --- a/pcapkit/protocols/schema/internet/hip.py +++ b/pcapkit/protocols/schema/internet/hip.py @@ -583,7 +583,7 @@ class R1CounterParameter(Parameter, code=[Enum_Parameter.R1_Counter, padding: 'bytes' = PaddingField(length=parameter_padding_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', counter: 'int') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', counter: 'int') -> 'None': ... @schema_final @@ -832,7 +832,7 @@ class NATTraversalModeParameter(Parameter, code=Enum_Parameter.NAT_TRAVERSAL_MOD padding: 'bytes' = PaddingField(length=parameter_padding_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', modes: 'list[Enum_NATTraversal]') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', modes: 'list[Enum_NATTraversal]') -> 'None': ... @schema_final @@ -941,7 +941,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': #: ``__init__``. cipher: 'Enum_Cipher' - def __init__(self, type: 'Enum_Parameter', len: 'int', + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', iv: 'Optional[bytes]', data: 'bytes') -> 'None': ... @@ -1069,7 +1069,7 @@ class NotificationParameter(Parameter, code=Enum_Parameter.NOTIFICATION): padding: 'bytes' = PaddingField(length=parameter_padding_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', msg_type: 'Enum_NotifyMessage', msg: 'bytes') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', msg_type: 'Enum_NotifyMessage', msg: 'bytes') -> 'None': ... @schema_final @@ -1174,7 +1174,7 @@ class RegFromParameter(Parameter, code=Enum_Parameter.REG_FROM): address: 'IPv6Address' = IPv6AddressField() if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', port: 'int', protocol: 'Enum_TransType', address: 'IPv6Address | bytes | int | str') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', port: 'int', protocol: 'Enum_TransType', address: 'IPv6Address | bytes | int | str') -> 'None': ... @schema_final @@ -1221,7 +1221,7 @@ class ESPTransformParameter(Parameter, code=Enum_Parameter.ESP_TRANSFORM): padding: 'bytes' = PaddingField(length=parameter_padding_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', suites: 'list[Enum_ESPTransformSuite]') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', suites: 'list[Enum_ESPTransformSuite]') -> 'None': ... @schema_final @@ -1269,7 +1269,7 @@ class PayloadMICParameter(Parameter, code=Enum_Parameter.PAYLOAD_MIC): padding: 'bytes' = PaddingField(length=parameter_padding_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', next: 'Enum_TransType', payload: 'bytes', mic: 'bytes') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', next: 'Enum_TransType', payload: 'bytes', mic: 'bytes') -> 'None': ... @schema_final @@ -1319,7 +1319,7 @@ class RouteDstParameter(Parameter, code=Enum_Parameter.ROUTE_DST): padding: 'bytes' = PaddingField(length=parameter_padding_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', flags: 'RouteFlags', hit: 'list[str | int | bytes | IPv6Address]') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', flags: 'RouteFlags', hit: 'list[str | int | bytes | IPv6Address]') -> 'None': ... @schema_final @@ -1436,7 +1436,7 @@ class RelayFromParameter(Parameter, code=Enum_Parameter.RELAY_FROM): address: 'IPv6Address' = IPv6AddressField() if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', port: 'int', protocol: 'Enum_TransType', address: 'str | bytes | int | IPv6Address') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', port: 'int', protocol: 'Enum_TransType', address: 'str | bytes | int | IPv6Address') -> 'None': ... @schema_final @@ -1453,7 +1453,7 @@ class RelayToParameter(Parameter, code=Enum_Parameter.RELAY_TO): address: 'IPv6Address' = IPv6AddressField() if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', port: 'int', protocol: 'Enum_TransType', address: 'str | bytes | int | IPv6Address') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', port: 'int', protocol: 'Enum_TransType', address: 'str | bytes | int | IPv6Address') -> 'None': ... @schema_final @@ -1468,7 +1468,7 @@ class OverlayTTLParameter(Parameter, code=Enum_Parameter.OVERLAY_TTL): padding: 'bytes' = PaddingField(length=parameter_padding_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', ttl: 'int') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', ttl: 'int') -> 'None': ... @schema_final @@ -1492,7 +1492,7 @@ class RouteViaParameter(Parameter, code=Enum_Parameter.ROUTE_VIA): padding: 'bytes' = PaddingField(length=parameter_padding_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', flags: 'RouteFlags', hit: 'list[str | bytes | int | IPv6Address]') -> 'None': ... + def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'bytes', flags: 'RouteFlags', hit: 'list[str | bytes | int | IPv6Address]') -> 'None': ... @schema_final diff --git a/pcapkit/protocols/schema/internet/hopopt.py b/pcapkit/protocols/schema/internet/hopopt.py index d8e7c03796..735ef0662e 100644 --- a/pcapkit/protocols/schema/internet/hopopt.py +++ b/pcapkit/protocols/schema/internet/hopopt.py @@ -427,7 +427,7 @@ class PadOption(Option, code=[Enum_Option.Pad1, pad: 'bytes' = PaddingField(length=pad_opt_data_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', len: 'int') -> 'None': ... + def __init__(self, type: 'Enum_Option', len: 'int', pad: 'bytes') -> 'None': ... @schema_final @@ -694,7 +694,7 @@ class QuickStartReportOption(QuickStartOption, code=Enum_QSFunction.Report_of_Ap if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', len: 'int', flags: 'QuickStartFlags', - nonce: 'QSNonce') -> 'None': ... + reserved: 'bytes', nonce: 'QSNonce') -> 'None': ... @schema_final diff --git a/pcapkit/protocols/schema/internet/ipv6_frag.py b/pcapkit/protocols/schema/internet/ipv6_frag.py index c1c7c790d5..d2db3232b9 100644 --- a/pcapkit/protocols/schema/internet/ipv6_frag.py +++ b/pcapkit/protocols/schema/internet/ipv6_frag.py @@ -48,4 +48,4 @@ class IPv6_Frag(Schema): payload: 'bytes' = PayloadField() if TYPE_CHECKING: - def __init__(self, next:'Enum_TransType', flags: 'Flags', id: 'int', payload: 'bytes | Schema | ProtocolBase') -> 'None': ... + def __init__(self, next:'Enum_TransType', reserved: 'bytes', flags: 'Flags', id: 'int', payload: 'bytes | Schema | ProtocolBase') -> 'None': ... diff --git a/pcapkit/protocols/schema/internet/ipv6_opts.py b/pcapkit/protocols/schema/internet/ipv6_opts.py index ba43afa4a3..a21d2c0588 100644 --- a/pcapkit/protocols/schema/internet/ipv6_opts.py +++ b/pcapkit/protocols/schema/internet/ipv6_opts.py @@ -427,7 +427,7 @@ class PadOption(Option, code=[Enum_Option.Pad1, pad: 'bytes' = PaddingField(length=pad_opt_data_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', len: 'int') -> 'None': ... + def __init__(self, type: 'Enum_Option', len: 'int', pad: 'bytes') -> 'None': ... @schema_final @@ -699,7 +699,7 @@ class QuickStartReportOption(QuickStartOption, code=Enum_QSFunction.Report_of_Ap if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', len: 'int', flags: 'QuickStartFlags', - nonce: 'QSNonce') -> 'None': ... + reserved: 'bytes', nonce: 'QSNonce') -> 'None': ... @schema_final diff --git a/pcapkit/protocols/schema/internet/ipv6_route.py b/pcapkit/protocols/schema/internet/ipv6_route.py index 118245fe04..0027c8d7eb 100644 --- a/pcapkit/protocols/schema/internet/ipv6_route.py +++ b/pcapkit/protocols/schema/internet/ipv6_route.py @@ -167,7 +167,7 @@ class SourceRoute(RoutingType, code=Enum_Routing.Source_Route): ) if TYPE_CHECKING: - def __init__(self, ip: 'list[IPv6Address | str | int | bytes]') -> 'None': ... + def __init__(self, reserved: 'bytes', ip: 'list[IPv6Address | str | int | bytes]') -> 'None': ... @schema_final @@ -180,7 +180,7 @@ class Type2(RoutingType, code=Enum_Routing.Type_2_Routing_Header): ip: 'IPv6Address' = IPv6AddressField() if TYPE_CHECKING: - def __init__(self, ip: 'IPv6Address | str | int | bytes') -> 'None': ... + def __init__(self, reserved: 'bytes', ip: 'IPv6Address | str | int | bytes') -> 'None': ... @schema_final @@ -312,4 +312,4 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': ip: 'list[IPv6Address | bytes]' def __init__(self, cmpr: 'CmprInfo', pad: 'PadInfo', - addresses: 'list[bytes]') -> 'None': ... + addresses: 'list[bytes]', padding: 'bytes') -> 'None': ... diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py index a2a2fd3482..f6b93a304b 100644 --- a/pcapkit/protocols/schema/internet/mh.py +++ b/pcapkit/protocols/schema/internet/mh.py @@ -675,7 +675,7 @@ class PadOption(Option, code=[Enum_Option.Pad1, data: 'bytes' = PaddingField(length=pad_opt_data_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int') -> 'None': ... + def __init__(self, type: 'Enum_Option', length: 'int', data: 'bytes') -> 'None': ... @schema_final @@ -1571,7 +1571,7 @@ class PadFlowIdentificationSuboption(FlowIdentificationSuboption, data: 'bytes' = PaddingField(length=pad_subopt_data_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_FlowIDSuboption', length: 'int') -> 'None': ... + def __init__(self, type: 'Enum_FlowIDSuboption', length: 'int', data: 'bytes') -> 'None': ... @schema_final diff --git a/pcapkit/protocols/schema/schema.py b/pcapkit/protocols/schema/schema.py index 0c851c2b08..e270a42b6c 100644 --- a/pcapkit/protocols/schema/schema.py +++ b/pcapkit/protocols/schema/schema.py @@ -862,7 +862,10 @@ class will consider negative value as a placeholder. continue if isinstance(field, PaddingField): - self.__buffer__[field.name] = bytes(field.length) + # NOTE: the octets as captured, or zeros when there are none + # (:issue:`1223`) -- see :meth:`PaddingField.pack + # `. + self.__buffer__[field.name] = field.pack(data, packet) continue if isinstance(field, ConditionalField): diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index 683fedaae4..2003ae6a94 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -161,12 +161,13 @@ if TYPE_CHECKING: from ipaddress import IPv4Address, IPv6Address - from typing import Any, Callable, DefaultDict, Optional, Sequence, Type + from typing import Any, Callable, DefaultDict, Optional, Sequence, Type, TypeVar from mypy_extensions import DefaultArg, KwArg, NamedArg from typing_extensions import Literal from pcapkit.const.reg.apptype import AppType as Enum_AppType + from pcapkit.protocols.data.data import Data from pcapkit.protocols.data.transport.sctp import Chunk as Data_Chunk from pcapkit.protocols.data.transport.sctp import ErrorCause as Data_ErrorCause from pcapkit.protocols.data.transport.sctp import Parameter as Data_Parameter @@ -179,6 +180,9 @@ Parameters = OrderedMultiDict[Enum_Parameter, Data_Parameter] Causes = OrderedMultiDict[Enum_CauseCode, Data_ErrorCause] + Data_T = TypeVar('Data_T', bound=Data) + Schema_T = TypeVar('Schema_T', bound=Schema) + ChunkParser = Callable[[Schema_Chunk, NamedArg(Chunks, 'chunks')], Data_Chunk] ChunkConstructor = Callable[[Enum_Chunk, DefaultArg(Optional[Data_Chunk]), KwArg(Any)], Schema_Chunk] @@ -205,6 +209,42 @@ del _index, _crc +def _keep_padding(schema: 'Schema', data: 'Data_T') -> 'Data_T': + """Keep a chunk's, parameter's or cause's non-zero trailing padding. + + Args: + schema: Parsed schema. + data: Data model read from ``schema``. + + 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. + + """ + padding = schema.__dict__.get('padding') + if isinstance(padding, bytes) and any(padding): + data.__update__([('padding', padding)]) + return data + + +def _restore_padding(data: 'Optional[Data]', schema: 'Schema_T') -> 'Schema_T': + """Write back the trailing padding :func:`_keep_padding` kept, if any. + + Args: + data: Data model the schema was made from, if any. + schema: Constructed schema. + + Returns: + ``schema``, with its padding set from ``data``. + + """ + padding = getattr(data, 'padding', None) + if isinstance(padding, bytes) and 'padding' in schema.__fields__: + schema.padding = padding + return schema + + class SCTP(Transport[Data_SCTP, Schema_SCTP], schema=Schema_SCTP, data=Data_SCTP): """This class implements Stream Control Transmission Protocol. @@ -891,7 +931,7 @@ def _read_sctp_chunks(self) -> 'Chunks': getattr(self, meth_name, self._read_chunk_donone)) else: meth = name[0] - chunks.add(code, meth(schema, chunks=chunks)) + chunks.add(code, _keep_padding(schema, meth(schema, chunks=chunks))) return chunks def _make_sctp_chunks(self, chunks: 'list[Schema_Chunk | tuple[Enum_Chunk, dict[str, Any]] | bytes] | Chunks') -> 'list[Schema_Chunk | bytes]': # pylint: disable=line-too-long @@ -950,7 +990,7 @@ def _make_sctp_chunk(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Chunk]' = N getattr(self, meth_name, self._make_chunk_donone)) else: meth = name[1] - return meth(code, chunk, **kwargs) + return _restore_padding(chunk, meth(code, chunk, **kwargs)) def _read_sctp_parameters(self, schemas: 'list[Schema_Parameter]') -> 'Parameters': """Read SCTP chunk parameter list. @@ -974,7 +1014,7 @@ def _read_sctp_parameters(self, schemas: 'list[Schema_Parameter]') -> 'Parameter getattr(self, meth_name, self._read_param_donone)) else: meth = name[0] - parameters.add(code, meth(schema, parameters=parameters)) + parameters.add(code, _keep_padding(schema, meth(schema, parameters=parameters))) return parameters def _make_sctp_parameters(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parameter, dict[str, Any]] | bytes] | Parameters') -> 'list[Schema_Parameter | bytes]': # pylint: disable=line-too-long @@ -1022,7 +1062,7 @@ def _make_sctp_parameter(self, code: 'Enum_Parameter', parameter: 'Optional[Data getattr(self, meth_name, self._make_param_donone)) else: meth = name[1] - return meth(code, parameter, **kwargs) + return _restore_padding(parameter, meth(code, parameter, **kwargs)) def _read_sctp_causes(self, schemas: 'list[Schema_ErrorCause]') -> 'Causes': """Read SCTP error cause list. @@ -1046,7 +1086,7 @@ def _read_sctp_causes(self, schemas: 'list[Schema_ErrorCause]') -> 'Causes': getattr(self, meth_name, self._read_cause_donone)) else: meth = name[0] - causes.add(code, meth(schema, causes=causes)) + causes.add(code, _keep_padding(schema, meth(schema, causes=causes))) return causes def _make_sctp_causes(self, causes: 'list[Schema_ErrorCause | tuple[Enum_CauseCode, dict[str, Any]] | bytes] | Causes') -> 'list[Schema_ErrorCause | bytes]': # pylint: disable=line-too-long @@ -1094,7 +1134,7 @@ def _make_sctp_cause(self, code: 'Enum_CauseCode', cause: 'Optional[Data_ErrorCa getattr(self, meth_name, self._make_cause_donone)) else: meth = name[1] - return meth(code, cause, **kwargs) + return _restore_padding(cause, meth(code, cause, **kwargs)) def _make_nested_length(self, base: 'int', items: 'Sequence[Schema_Parameter | Schema_ErrorCause | bytes]', # pylint: disable=line-too-long declared: 'Optional[int]' = None) -> 'int': diff --git a/tests/protocols/application/test_http_unit.py b/tests/protocols/application/test_http_unit.py index 34d19ad9f9..5228d0b6dd 100644 --- a/tests/protocols/application/test_http_unit.py +++ b/tests/protocols/application/test_http_unit.py @@ -1925,13 +1925,14 @@ def test_httpv2_frame_readers_cover_successful_frames(self) -> None: warn.assert_not_called() data = proto._read_http_data( - http2_schema(pad_len=2, data=b'data', __flags__=0x09), + http2_schema(pad_len=2, data=b'data', padding=b'\xee\xff', __flags__=0x09), header=http2_header(7, Frame.DATA, flags=0x09, sid=1), ) self.assertTrue(data.flags.to_dict()['END_STREAM']) self.assertTrue(data.flags.to_dict()['PADDED']) self.assertEqual(data.to_dict()['pad_len'], 2) self.assertEqual(data.to_dict()['data'], b'data') + self.assertEqual(data.to_dict()['padding'], b'\xee\xff') headers = proto._read_http_headers( http2_schema( @@ -1939,6 +1940,7 @@ def test_httpv2_frame_readers_cover_successful_frames(self) -> None: stream_dep={'exclusive': 1, 'sid': 7}, weight=4, fragment=b'headers', + padding=b'\x00', __flags__=0x2D, ), header=http2_header(14, Frame.HEADERS, flags=0x2D, sid=1), @@ -1977,7 +1979,8 @@ def test_httpv2_frame_readers_cover_successful_frames(self) -> None: self.assertEqual(settings.settings[Setting.ENABLE_PUSH], 0) push = proto._read_http_push_promise( - http2_schema(pad_len=1, stream={'reserved': 0, 'sid': 11}, fragment=b'push', __flags__=0x0C), + http2_schema(pad_len=1, stream={'reserved': 0, 'sid': 11}, fragment=b'push', padding=b'\x00', + __flags__=0x0C), header=http2_header(10, Frame.PUSH_PROMISE, flags=0x0C, sid=1), ) self.assertTrue(push.flags.to_dict()['END_HEADERS']) diff --git a/tests/protocols/application/test_ospf_unit.py b/tests/protocols/application/test_ospf_unit.py index 111dcdac5c..0d02f058bf 100644 --- a/tests/protocols/application/test_ospf_unit.py +++ b/tests/protocols/application/test_ospf_unit.py @@ -134,7 +134,7 @@ def test_ospf_properties_read_make_and_auth_helpers(self) -> None: self.assertEqual(schema.length, 28) self.assertEqual(schema.auth_data, b'\x01' * 8) - crypto_data_model = DataCryptoAuth(key_id=2, len=20, seq=100) + crypto_data_model = DataCryptoAuth(reserved=b'\x00\x00', key_id=2, len=20, seq=100) crypto_schema_from_data = maker.make( auth_type=Authentication.Cryptographic_authentication, auth_data=crypto_data_model, diff --git a/tests/protocols/internet/test_hip_unit.py b/tests/protocols/internet/test_hip_unit.py index e33b6edce4..0ef570c145 100644 --- a/tests/protocols/internet/test_hip_unit.py +++ b/tests/protocols/internet/test_hip_unit.py @@ -1394,7 +1394,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> Parameter.R1_COUNTER, hip_data.R1CounterParameter(type=Parameter.R1_COUNTER, critical=False, length=16, - counter=2), + reserved=bytes(4), counter=2), version=2, ).counter, 2) @@ -1505,7 +1505,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_nat_traversal_mode( Parameter.NAT_TRAVERSAL_MODE, hip_data.NATTraversalModeParameter(type=Parameter.NAT_TRAVERSAL_MODE, - critical=False, length=8, + critical=False, reserved=bytes(2), length=8, mode_id=(NATTraversal.UDP_ENCAPSULATION,)), version=2, ).modes, (NATTraversal.UDP_ENCAPSULATION,)) @@ -1523,7 +1523,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_encrypted( Parameter.ENCRYPTED, hip_data.EncryptedParameter(type=Parameter.ENCRYPTED, - critical=False, length=8, + critical=False, reserved=bytes(4), length=8, cipher=Cipher.NULL_ENCRYPT, iv=b'iv', data=b'data'), version=2, @@ -1580,7 +1580,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_notification( Parameter.NOTIFICATION, hip_data.NotificationParameter(type=Parameter.NOTIFICATION, - critical=False, length=8, + critical=False, reserved=bytes(2), length=8, msg_type=NotifyMessage.INVALID_SYNTAX, msg=b'bad'), version=2, @@ -1635,7 +1635,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> version=2).reg_failed, []) self.assertEqual(proto._make_param_reg_from( Parameter.REG_FROM, - hip_data.RegFromParameter(type=Parameter.REG_FROM, critical=False, + hip_data.RegFromParameter(type=Parameter.REG_FROM, critical=False, reserved=bytes(1), length=24, port=10500, protocol=TransType.UDP, address=ip_address('2001:db8::10')), @@ -1662,7 +1662,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_esp_transform( Parameter.ESP_TRANSFORM, hip_data.ESPTransformParameter(type=Parameter.ESP_TRANSFORM, - critical=False, length=8, + critical=False, reserved=bytes(2), length=8, suite_id=(ESPTransformSuite.AES_128_CBC_with_HMAC_SHA1,)), version=2, ).suites, (ESPTransformSuite.AES_128_CBC_with_HMAC_SHA1,)) @@ -1687,7 +1687,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_payload_mic( Parameter.PAYLOAD_MIC, hip_data.PayloadMICParameter(type=Parameter.PAYLOAD_MIC, - critical=False, length=16, + critical=False, reserved=bytes(3), length=16, next=TransType.TCP, payload=b'payload', mic=b'mic'), version=2, @@ -1708,7 +1708,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertTrue(proto._make_param_route_dst( Parameter.ROUTE_DST, hip_data.RouteDstParameter(type=Parameter.ROUTE_DST, - critical=False, length=24, + critical=False, reserved=bytes(2), length=24, flags=flags, hit=(ip_address('2001:db8::20'),)), version=2, @@ -1770,7 +1770,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_relay_from( Parameter.RELAY_FROM, hip_data.RelayFromParameter(type=Parameter.RELAY_FROM, - critical=False, length=24, + critical=False, reserved=bytes(1), length=24, port=10501, protocol=TransType.UDP, address=ip_address('2001:db8::30')), version=2, @@ -1778,7 +1778,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_relay_to( Parameter.RELAY_TO, hip_data.RelayToParameter(type=Parameter.RELAY_TO, - critical=False, length=24, + critical=False, reserved=bytes(1), length=24, port=10502, protocol=TransType.UDP, address=ip_address('2001:db8::31')), version=2, @@ -1786,13 +1786,13 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_overlay_ttl( Parameter.OVERLAY_TTL, hip_data.OverlayTTLParameter(type=Parameter.OVERLAY_TTL, - critical=False, length=8, ttl=dt1), + critical=False, reserved=bytes(2), length=8, ttl=dt1), version=2, ).ttl, 1) self.assertTrue(proto._make_param_route_via( Parameter.ROUTE_VIA, hip_data.RouteViaParameter(type=Parameter.ROUTE_VIA, - critical=False, length=24, + critical=False, reserved=bytes(2), length=24, flags=flags, hit=(ip_address('2001:db8::40'),)), version=2, diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 9bea01030a..c394930b78 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -165,6 +165,7 @@ def test_ipv6_frag_index_length_and_make_data(self) -> None: data = DummyDict( next=TransType.TCP, offset=16, + reserved_octet=b'\x00', reserved=0, mf=True, id=99, @@ -348,7 +349,7 @@ def test_ipv6_route_readers_and_constructors_cover_registered_types(self) -> Non self.assertEqual(str(proto._make_data_type_src( Routing.Source_Route, route_data.SourceRoute(next=TransType.TCP, length=24, type=Routing.Source_Route, - seg_left=1, ip=(ip_address('2001:db8::5'),)), + seg_left=1, reserved=bytes(4), ip=(ip_address('2001:db8::5'),)), ).ip[0]), '2001:db8::5') self.assertEqual(str(proto._make_data_type_2( Routing.Type_2_Routing_Header, @@ -357,7 +358,7 @@ def test_ipv6_route_readers_and_constructors_cover_registered_types(self) -> Non self.assertEqual(str(proto._make_data_type_2( Routing.Type_2_Routing_Header, route_data.Type2(next=TransType.TCP, length=24, type=Routing.Type_2_Routing_Header, - seg_left=1, ip=ip_address('2001:db8::7')), + seg_left=1, reserved=bytes(4), ip=ip_address('2001:db8::7')), ).ip), '2001:db8::7') rpl_plain = proto._make_data_type_rpl( @@ -375,7 +376,7 @@ def test_ipv6_route_readers_and_constructors_cover_registered_types(self) -> Non Routing.RPL_Source_Route_Header, route_data.RPL(next=TransType.TCP, length=16, type=Routing.RPL_Source_Route_Header, seg_left=1, cmpr_i=1, cmpr_e=2, pad=0, reserved=0, - ip=(ip_address('2001:db8::9'),)), + ip=(ip_address('2001:db8::9'),), padding=b''), ) self.assertEqual(rpl_from_data.cmpr['cmpr_i'], 1) @@ -561,7 +562,7 @@ def test_ipv6_route_read_make_registry_and_property_edges(self) -> None: type=Routing.Type_2_Routing_Header, data=route_data.Type2(next=TransType.TCP, length=24, type=Routing.Type_2_Routing_Header, - seg_left=1, + seg_left=1, reserved=bytes(4), ip=ip_address('2001:db8::3')), ) self.assertEqual(made_model.type, Routing.Type_2_Routing_Header) @@ -852,6 +853,7 @@ def _assert_option_data_model_constructor_edges(self, protocol_cls: type) -> Non rate=80, nonce=4, reserved=0, + unused=b'\x00', **base), ).nonce['nonce'], 4) with self.assertRaises(ProtocolError): @@ -1087,7 +1089,7 @@ def _assert_option_constructors_cover_branchy_values(self, protocol_cls: type) - self.assertEqual(no_pad_options, [bytes([Option.CALIPSO]) + b'12345']) dict_options, dict_length = make_options(OrderedMultiDict([ - (Option.Pad1, data.PadOption(type=Option.Pad1, length=1, **base)), + (Option.Pad1, data.PadOption(type=Option.Pad1, length=1, pad=b'', **base)), (Option.Tunnel_Encapsulation_Limit, data.TunnelEncapsulationLimitOption(type=Option.Tunnel_Encapsulation_Limit, length=3, limit=3, **base)), diff --git a/tests/protocols/internet/test_mh_padn_contents_unit.py b/tests/protocols/internet/test_mh_padn_contents_unit.py new file mode 100644 index 0000000000..78b9344000 --- /dev/null +++ b/tests/protocols/internet/test_mh_padn_contents_unit.py @@ -0,0 +1,72 @@ +# -*- coding: utf-8 -*- +"""MH ``PadN`` contents rebuild as captured. + +GitHub issue #1329, its ``PadN`` remainder: the contents of a ``PadN`` +mobility option, and of a Flow Identification ``PadN`` sub-option, were held by +a :class:`~pcapkit.corekit.fields.strings.PaddingField` that packed zeros, so +``0106aabbccddeeff`` came back as ``0106000000000000``. The contents are now +kept on the data model (``data``) and written back; a fresh build still writes +zeros. The root cause is #1223. + +Every case builds its own octets in memory and reads no capture. :class:`MH` is +imported inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import io +import unittest + +from tests._support import reimport_once_per_class + + +class TestMHPadNContents(unittest.TestCase): + """Pin the wire form of MH ``PadN`` contents across rebuilds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_padn_option_contents_rebuild_from_data(self) -> None: + from pcapkit.protocols.internet.mh import MH + + # BRR, then a PadN of Option Length 6 with non-zero contents + raw = bytes.fromhex('3b01000000000000' '0106aabbccddeeff') + parsed = MH(io.BytesIO(raw), len(raw), extension=True) + option = next(iter(parsed.info.options.values())) + self.assertEqual(option.data, bytes.fromhex('aabbccddeeff')) + self.assertEqual(bytes(MH.from_data(parsed.info).data).hex(), raw.hex()) + + def test_padn_option_make_writes_zeros_unless_given(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + + proto = object.__new__(MH) + self.assertEqual(bytes(proto._make_opt_pad(Option.PadN, length=2)).hex(), '01020000') + self.assertEqual(bytes(proto._make_opt_pad(Option.PadN, length=2, data=b'\xaa\xbb')).hex(), + '0102aabb') + + def test_padn_suboption_contents_round_trip(self) -> None: + from pcapkit.const.mh.flow_id_suboption import FlowIDSuboption + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.mh import MH + + def build(pad: 'dict[str, object]') -> 'bytes': + args = {'fid': 7, 'fid_pri': 2, 'status': 0, + 'suboptions': [(FlowIDSuboption.PadN, pad)]} + return bytes(MH(next=TransType.UDP, chksum=b'\x12\x34', + type=Packet.Binding_Refresh_Request, + data={'options': [(Option.Flow_Identification_Mobility_Option, args)]})) + + zeros = build({'length': 2}) + raw = build({'length': 2, 'data': b'\xaa\xbb'}) + self.assertEqual(raw.replace(b'\x01\x02\xaa\xbb', b'\x01\x02\x00\x00'), zeros) + self.assertNotEqual(raw, zeros) + + parsed = MH(io.BytesIO(raw), len(raw), extension=True).info + rebuilt = bytes(MH(next=parsed.next, type=parsed.type, chksum=parsed.chksum, data=parsed)) + self.assertEqual(rebuilt.hex(), raw.hex()) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 9edb916ee8..ddf9e9492c 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -768,7 +768,7 @@ def make_extension(code: CGAExtension, extension: data.UnknownExtension | None = # asserting the two-octet overshoot. remade = proto._make_opt_pad( Option.Pad1, - data.PadOption(type=Option.PadN, length=4), + data.PadOption(type=Option.PadN, length=4, data=b'\x00\x00'), ) self.assertEqual(remade.length, 2) self.assertEqual(len(remade.pack()), 4) @@ -2029,7 +2029,8 @@ def test_mh_padding_option_data_model_round_trips(self) -> None: proto = object.__new__(MH) for code, total in ((Option.Pad1, 1), (Option.PadN, 3), (Option.PadN, 6)): with self.subTest(option=code.name, length=total): - made = proto._make_opt_pad(code, PadOption(type=code, length=total)) + made = proto._make_opt_pad(code, PadOption(type=code, length=total, + data=bytes(max(total - 2, 0)))) self.assertEqual(made.type, code) self.assertEqual(len(made.pack()), total) diff --git a/tests/protocols/test_padding_kept_as_captured_unit.py b/tests/protocols/test_padding_kept_as_captured_unit.py new file mode 100644 index 0000000000..dff56305c1 --- /dev/null +++ b/tests/protocols/test_padding_kept_as_captured_unit.py @@ -0,0 +1,207 @@ +# -*- coding: utf-8 -*- +"""Padding and reserved octets held by a ``PaddingField`` rebuild as captured. + +GitHub issue #1223. :class:`~pcapkit.corekit.fields.strings.PaddingField` packed +zeros whatever it had read, so non-zero padding or reserved octets parsed +silently and came back zeroed. The field now packs the octets it was given, +and zeros when it has none, which is still what a freshly built packet writes. +Each protocol below carries the octets through its reader, data model and +``make()``; the HIP and SCTP trailing alignment padding is kept on the data +model only when it is not all zeros. + +Every case builds its own octets in memory and reads no capture. Classes are +imported inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import io +import unittest + +from tests._support import reimport_once_per_class + +#: An IPv6 address, for the routing headers. +ADDR = '20010db8000000000000000000000001' + +#: ``(module, class, extension)`` -> wire octets that must rebuild unchanged. +FRAMES = { + 'HOPOPT PadN data': ('internet.hopopt', 'HOPOPT', True, '3b000104aabbccdd'), + 'IPv6-Opts PadN data': ('internet.ipv6_opts', 'IPv6_Opts', True, '3b000104aabbccdd'), + 'HOPOPT QS report Not Used octet': ('internet.hopopt', 'HOPOPT', True, + '3b01' '260680aa00000000' '010400000000'), + 'IPv6-Opts QS report Not Used octet': ('internet.ipv6_opts', 'IPv6_Opts', True, + '3b01' '260680aa00000000' '010400000000'), + 'IPv6-Frag reserved octet': ('internet.ipv6_frag', 'IPv6_Frag', True, '3baa000000000001'), + 'IPv6-Route Source Route reserved': ('internet.ipv6_route', 'IPv6_Route', True, + '3b020001aabbccdd' + ADDR), + 'IPv6-Route Type 2 reserved': ('internet.ipv6_route', 'IPv6_Route', True, + '3b020201aabbccdd' + ADDR), + # CmprI 15, CmprE 15, Pad 7: one 1-octet address then 7 pad octets + 'IPv6-Route RPL padding': ('internet.ipv6_route', 'IPv6_Route', True, + '3b010301ff700000' '01eeeeeeeeeeeeee'), + 'HTTP/2 DATA padding': ('application.httpv2', 'HTTP', False, '00000600080000000102616263eeff'), + 'HTTP/2 HEADERS padding': ('application.httpv2', 'HTTP', False, '0000040108000000010282eeff'), + 'HTTP/2 PUSH_PROMISE padding': ('application.httpv2', 'HTTP', False, + '000008050800000001020000000282eeff'), + 'OSPF cryptographic authentication reserved': ( + 'application.ospf', 'OSPF', False, + '0201002c' 'c0a8000100000000' '0000' '0002' 'aabb011000000007' + 'ffffff00000a020100000028c0a8000100000000' + '00' * 16), + 'TCP octets after EOOL': ('transport.tcp', 'TCP', False, + '0050005000000001000000007002ffff00000000' '020405b400aabbcc'), + 'SCTP DATA chunk padding': ('transport.sctp', 'SCTP', False, + '138813891122334400000000' '00030011' '00000001' '0000' '0000' + '00000000' '61ffffff'), + 'SCTP parameter padding': ('transport.sctp', 'SCTP', False, + '138813891122334400000000' '0100001b' '00000001' '0000ffff' + '00010001' '00000001' '80010007616263ee'), + 'SCTP error cause padding': ('transport.sctp', 'SCTP', False, + '138813891122334400000000' '06000009' '00ff0005aaeeeeee'), +} + +#: HIP parameters whose reserved octets are ``ff`` (the 12 sites listed on +#: #1223), and one whose trailing padding is not zero. +HIP_PARAMS = { + 'R1_COUNTER': '0081000cffffffff00000000aabbccdd', + 'NAT_TRAVERSAL_MODE': '02600002ffff0000', + 'ENCRYPTED': '02810004ffffffff', + 'NOTIFICATION': '03400004ffff4000', + 'REG_FROM': '03b60014000011ff' + '00' * 16, + 'ESP_TRANSFORM': '0fff0002ffff0000', + 'PAYLOAD_MIC': '11e1000811ffffff0000000000000000', + 'ROUTE_DST': '11f900140000ffff' + ADDR, + 'RELAY_FROM': 'f9fe0014000011ff' + '00' * 16, + 'RELAY_TO': 'fa020014000011ff' + '00' * 16, + 'OVERLAY_TTL': 'fa0b00040000ffff', + 'ROUTE_VIA': 'fa1100140000ffff' + ADDR, + 'NOTIFICATION trailing padding': '03400005000040006100000000eeeeee', +} + + +def hip_packet(param: 'str') -> 'bytes': + """A HIP packet (next header 59, HIPv2 I1) carrying one parameter.""" + body = bytes.fromhex(param) + length = (40 + len(body)) // 8 - 1 + return bytes([0x3b, length, 0x01, 0x21]) + bytes(36) + body + + +class TestPaddingKeptAsCaptured(unittest.TestCase): + """Pin the wire form of padding and reserved octets across rebuilds.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_padding_field_packs_what_it_was_given(self) -> None: + from pcapkit.corekit.fields.strings import PaddingField + from pcapkit.protocols.schema.schema import Schema, schema_final + + @schema_final + class Padded(Schema): + pad: 'bytes' = PaddingField(length=3) + + self.assertEqual(bytes(Padded.unpack(b'\xaa\xbb\xcc')), b'\xaa\xbb\xcc') + self.assertEqual(bytes(Padded()), bytes(3)) + self.assertEqual(bytes(Padded(pad=b'\x01\x02\x03')), b'\x01\x02\x03') + # a value of another width is zero-filled or truncated, as ``struct``'s + # ``s`` format does; an absent or empty one packs as zeros + self.assertEqual(bytes(Padded(pad=b'\x01')), b'\x01\x00\x00') + self.assertEqual(bytes(Padded(pad=b'\x01\x02\x03\x04')), b'\x01\x02\x03') + self.assertEqual(bytes(Padded(pad=b'')), bytes(3)) + self.assertEqual(PaddingField(length=2).pack(None, {}), bytes(2)) + + def test_wrong_width_padding_is_zero_filled_or_truncated(self) -> None: + """A ``ConditionalField``-wrapped padding keeps ``main``'s ``struct`` semantics.""" + from pcapkit.const.http.frame import Frame + from pcapkit.protocols.application.httpv2 import HTTP + from pcapkit.protocols.link.l2tpv2 import L2TPv2 + + # the L2TPv2 ``padding`` contract: zero-filled or truncated to ``offset`` + # octets (bytes measured on ``main``) + for padding, wire in ((b'\xaa', '0202000000000003aa00007878'), + (b'\xaa\xbb\xcc\xdd\xee', '0202000000000003aabbcc7878'), + (None, '02020000000000030000007878')): + with self.subTest(l2tpv2=padding): + self.assertEqual(L2TPv2(offset=3, padding=padding, payload=b'xx').data.hex(), wire) + + for padding, wire in ((b'\xaa', '0000050008000000010378aa0000'), + (b'\xaa\xbb\xcc\xdd', '0000050008000000010378aabbcc'), + (b'', '0000050008000000010378000000')): + with self.subTest(http2=padding): + frame = {'pad_len': 3, 'data': b'x', 'padding': padding} + self.assertEqual(HTTP(type=Frame.DATA, sid=1, frame=frame).data.hex(), wire) + + def test_makers_tolerate_a_data_model_without_the_new_fields(self) -> None: + """A stand-in lacking the new attributes rebuilds with zero padding.""" + from types import SimpleNamespace + + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.hopopt import HOPOPT + + proto = object.__new__(HOPOPT) + opt = SimpleNamespace(type=Option.PadN, length=6) + self.assertEqual(bytes(proto._make_opt_pad(Option.PadN, opt)).hex(), '010400000000') + + def test_from_data_rebuilds_byte_for_byte(self) -> None: + import importlib + + for name, (module, cls_name, extension, wire) in FRAMES.items(): + with self.subTest(name): + cls = getattr(importlib.import_module(f'pcapkit.protocols.{module}'), cls_name) + raw = bytes.fromhex(wire) + kwargs = {'extension': True} if extension else {} + parsed = cls(io.BytesIO(raw), len(raw), **kwargs) + self.assertEqual(bytes(cls.from_data(parsed.info).data).hex(), raw.hex()) + + def test_hip_from_data_rebuilds_byte_for_byte(self) -> None: + from pcapkit.protocols.internet.hip import HIP + + for name, param in HIP_PARAMS.items(): + with self.subTest(name): + raw = hip_packet(param) + parsed = HIP(io.BytesIO(raw), len(raw), extension=True) + self.assertEqual(bytes(HIP.from_data(parsed.info).data).hex(), raw.hex()) + + def test_zero_padding_is_not_recorded(self) -> None: + """HIP and SCTP alignment padding stays off the data model while zero.""" + from pcapkit.protocols.internet.hip import HIP + from pcapkit.protocols.transport.sctp import SCTP + + raw = hip_packet('03400005000040006100000000000000') + param = next(iter(HIP(io.BytesIO(raw), len(raw), extension=True).info.parameters.values())) + self.assertNotIn('padding', param) + raw = hip_packet(HIP_PARAMS['NOTIFICATION trailing padding']) + param = next(iter(HIP(io.BytesIO(raw), len(raw), extension=True).info.parameters.values())) + self.assertEqual(param.padding, b'\x00' * 4 + b'\xee' * 3) + + raw = bytes.fromhex(FRAMES['SCTP DATA chunk padding'][3]) + chunk = next(iter(SCTP(raw).info.chunks.values())) + self.assertEqual(chunk.padding, b'\xff' * 3) + chunk = next(iter(SCTP(raw[:-3] + bytes(3)).info.chunks.values())) + self.assertNotIn('padding', chunk) + + def test_make_writes_zeros_by_default(self) -> None: + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag + + proto = object.__new__(HOPOPT) + self.assertEqual(bytes(proto._make_opt_pad(Option.PadN, length=4)).hex(), '010400000000') + self.assertEqual(bytes(proto._make_opt_pad(Option.PadN, length=4, pad=b'\xaa' * 4)).hex(), + '0104aaaaaaaa') + self.assertEqual(IPv6_Frag(next=59, id=1, extension=True).data.hex(), '3b00000000000001') + self.assertEqual(IPv6_Frag(next=59, id=1, reserved_octet=b'\xaa', extension=True).data.hex(), + '3baa000000000001') + + def test_make_takes_the_hip_reserved_octets(self) -> None: + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.internet.hip import HIP + + base = {'next': 59, 'packet': 1, 'version': 2, 'checksum': b'\x00\x00', + 'controls_anonymous': False, 'shit': 0, 'rhit': 0, 'payload': b''} + params = [(Parameter.NOTIFICATION, {})] + self.assertEqual(HIP(parameters=params, extension=True, **base).data[44:46], b'\x00\x00') + params = [(Parameter.NOTIFICATION, {'reserved': b'\xff\xff'})] + self.assertEqual(HIP(parameters=params, extension=True, **base).data[44:46], b'\xff\xff') + + +if __name__ == '__main__': + unittest.main()