diff --git a/pcapkit/protocols/data/internet/ipv4.py b/pcapkit/protocols/data/internet/ipv4.py index b901090e47..458acc0103 100644 --- a/pcapkit/protocols/data/internet/ipv4.py +++ b/pcapkit/protocols/data/internet/ipv4.py @@ -79,13 +79,15 @@ def __new__(cls, *args: 'Any', **kwargs: 'Any') -> 'ToSField': class Flags(Data): """Data model for IPv4 Flags.""" + #: Reserved flag bit. + reserved: 'int' #: Don't fragment. df: 'bool' #: More fragments. mf: 'bool' if TYPE_CHECKING: - def __init__(self, df: 'bool', mf: 'bool') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, reserved: 'int', df: 'bool', mf: 'bool') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -120,6 +122,8 @@ class IPv4(Protocol): if TYPE_CHECKING: #: Options, present only when the header carries any. options: 'OrderedMultiDict[OptionNumber, Option]' + #: Octets after the End of Option List, present along with ``options``. + padding: 'bytes' def __init__(self, version: 'Literal[4]', hdr_len: 'int', tos: 'ToSField', len: 'int', id: 'int', flags: 'Flags', offset: 'int', ttl: 'timedelta', protocol: 'TransType', checksum: 'bytes', src: 'IPv4Address', dst: 'IPv4Address') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @@ -202,11 +206,13 @@ class LSROption(Option): #: Pointer. pointer: 'int' - #: Route. + #: Route, i.e. the address slots before the pointer. route: 'tuple[IPv4Address, ...]' + #: Address slots at or beyond the pointer. + remaining: 'tuple[IPv4Address, ...]' if TYPE_CHECKING: - def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', route: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', route: 'tuple[IPv4Address, ...]', remaining: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -245,11 +251,13 @@ class RROption(Option): #: Pointer. pointer: 'int' - #: Route. + #: Route, i.e. the address slots before the pointer. route: 'Optional[tuple[IPv4Address, ...]]' + #: Address slots at or beyond the pointer. + remaining: 'tuple[IPv4Address, ...]' if TYPE_CHECKING: - def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', route: 'Optional[tuple[IPv4Address, ...]]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', route: 'Optional[tuple[IPv4Address, ...]]', remaining: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -269,11 +277,13 @@ class SSROption(Option): #: Pointer. pointer: 'int' - #: Route. + #: Route, i.e. the address slots before the pointer. route: 'Optional[tuple[IPv4Address, ...]]' + #: Address slots at or beyond the pointer. + remaining: 'tuple[IPv4Address, ...]' if TYPE_CHECKING: - def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', route: 'Optional[tuple[IPv4Address, ...]]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', pointer: 'int', route: 'Optional[tuple[IPv4Address, ...]]', remaining: 'tuple[IPv4Address, ...]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final @@ -355,17 +365,23 @@ class QuickStartRequestOption(QSOption): ttl: 'timedelta' #: Nonce. nonce: 'int' + #: Reserved bits after the nonce. + reserved: 'int' if TYPE_CHECKING: - def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', func: 'QSFunction', rate: 'int', ttl: 'timedelta', nonce: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', func: 'QSFunction', rate: 'int', ttl: 'timedelta', nonce: 'int', reserved: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin @info_final class QuickStartReportOption(QSOption): """Data model for IPv4 Quick Start report of approved rate option.""" + #: Not-used octet, where a request carries its TTL. + unused: 'int' #: Nonce. nonce: 'int' + #: Reserved bits after the nonce. + reserved: 'int' if TYPE_CHECKING: - def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', func: 'QSFunction', rate: 'int', nonce: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin + def __init__(self, code: 'OptionNumber', length: 'int', type: 'OptionType', func: 'QSFunction', rate: 'int', unused: 'int', nonce: 'int', reserved: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long,redefined-builtin diff --git a/pcapkit/protocols/internet/ipv4.py b/pcapkit/protocols/internet/ipv4.py index b0323ea94d..f652962d87 100644 --- a/pcapkit/protocols/internet/ipv4.py +++ b/pcapkit/protocols/internet/ipv4.py @@ -304,6 +304,7 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu len=schema.length, id=schema.id, flags=Data_Flags( + reserved=schema.flags['reserved'], df=bool(schema.flags['df']), mf=bool(schema.flags['mf']), ), @@ -317,8 +318,12 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu _optl = ipv4.hdr_len - 20 if _optl: + options = self._read_ipv4_options(_optl) + _used = sum(opt.length for _, opt in options.items(multi=True)) ipv4.__update__([ - ('options', self._read_ipv4_options(_optl)), + ('options', options), + # octets after the End of Option List, as captured + ('padding', self._data[20 + _used:ipv4.hdr_len]), ]) # update packet info @@ -362,6 +367,7 @@ def make(self, tos_ecn_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long tos_ecn_reversed: 'bool' = False, id: 'int' = 0, + reserved: 'int' = 0, df: 'bool' = False, mf: 'bool' = False, offset: 'int' = 0, @@ -375,6 +381,7 @@ def make(self, dst: 'IPv4Address | str | int | bytes' = '0.0.0.0', # nosec: B104 options: 'Optional[list[Schema_Option | tuple[Enum_OptionNumber, dict[str, Any]] | bytes] | Option]' = None, # pylint: disable=line-too-long total_length: 'Optional[int]' = None, + padding: 'Optional[bytes]' = None, payload: 'bytes | ProtocolBase | Schema' = b'', **kwargs: 'Any') -> 'Schema_IPv4': """Make (construct) packet data. @@ -401,6 +408,7 @@ def make(self, tos_ecn_namespace: Namespace of ``tos_ecn``. tos_ecn_reversed: If the namespace of ``tos_ecn`` is reversed. id: Identification of the packet. + reserved: Reserved flag bit, i.e. the high-order bit of the flags. df: Don't fragment flag. mf: More fragments flag. offset: Fragment offset. @@ -415,6 +423,9 @@ def make(self, options: Options of the packet. total_length: Total length of the packet, header included; computed from the header and the payload when omitted. + padding: Octets after the End of Option List, written as is when + ``options`` is a container that carries an ``EOOL``, c.f. + :meth:`_make_ipv4_options`. payload: Payload of the packet. **kwargs: Arbitrary keyword arguments. @@ -438,7 +449,7 @@ def make(self, ttl_val = ttl if isinstance(ttl, int) else math.ceil(ttl.total_seconds()) if options is not None: - options_value, options_length = self._make_ipv4_options(options) + options_value, options_length = self._make_ipv4_options(options, padding) else: options_value, options_length = [], 0 @@ -467,6 +478,7 @@ def make(self, length=total_length, id=id, flags={ + 'reserved': reserved, 'df': df, 'mf': mf, 'offset': offset, @@ -549,6 +561,7 @@ def _make_data(cls, data: 'Data_IPv4') -> 'dict[str, Any]': # type: ignore[over 'tos_rel': data.tos.rel, 'tos_ecn': data.tos.ecn, 'id': data.id, + 'reserved': data.flags.reserved, 'df': data.flags.df, 'mf': data.flags.mf, 'offset': data.offset // 8, @@ -565,6 +578,10 @@ def _make_data(cls, data: 'Data_IPv4') -> 'dict[str, Any]': # type: ignore[over # to ``make`` to recompute, so that a truncated capture rebuilds # with the length that was on the wire (:issue:`1155`). 'total_length': data.len, + # NOTE: And the octets after the End of Option List, which no option + # records. The header length is not passed: it follows from the + # options and these octets, so edited options recompute it. + 'padding': getattr(data, 'padding', None), 'payload': cls._make_payload(data), } @@ -663,10 +680,10 @@ def _read_opt_unassigned(self, schema: 'Schema_UnassignedOption', *, options: 'O Parsed option data. Raises: - ProtocolError: If ``length`` is **LESS THAN** ``3``. + ProtocolError: If ``length`` is **LESS THAN** ``2``. """ - if schema.length < 3: + if schema.length < 2: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') opt = Data_UnassignedOption( @@ -756,6 +773,11 @@ def _read_opt_sec(self, schema: 'Schema_SECOption', *, options: 'Option') -> 'Da Raises: ProtocolError: If ``length`` is **LESS THAN** ``3``. + Notes: + Per :rfc:`1108#section-2.4`, the low-order bit of each protection + authority octet is the field termination indicator: ``0`` on the + final octet, ``1`` when more octets follow. + """ if schema.length < 3: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') @@ -770,11 +792,11 @@ def _read_opt_sec(self, schema: 'Schema_SECOption', *, options: 'Option') -> 'Da warn(f'{self.alias}: [OptNo {schema.type}] invalid format: unknown protection authority: {authority}', ProtocolWarning) flags.append(authority) - if byte & 0x01 == 1 and base < schema.length - 4: + if byte & 0x01 == 0 and base < schema.length - 4: #raise ProtocolError(f'{self.alias}: [OptNo {kind}] invalid format: remaining data') warn(f'{self.alias}: [OptNo {schema.type}] invalid format: remaining data', ProtocolWarning) - if schema.data[-1] & 0x01 == 0: + if schema.data[-1] & 0x01 == 1: warn(f'{self.alias}: [OptNo {schema.type}] invalid format: field termination indicator not set', ProtocolWarning) else: flags = [] @@ -821,7 +843,8 @@ def _read_opt_lsr(self, schema: 'Schema_LSROption', *, options: 'Option') -> 'Da type=self._read_ipv4_opt_type(schema.type), length=schema.length, pointer=schema.pointer, - route=tuple(schema.route), + route=tuple(schema.route[:(schema.pointer - 4) // 4]), + remaining=tuple(schema.route[(schema.pointer - 4) // 4:]), ) return opt @@ -940,7 +963,8 @@ def _read_opt_rr(self, schema: 'Schema_RROption', *, options: 'Option') -> 'Data type=self._read_ipv4_opt_type(schema.type), length=schema.length, pointer=schema.pointer, - route=tuple(schema.route), + route=tuple(schema.route[:(schema.pointer - 4) // 4]), + remaining=tuple(schema.route[(schema.pointer - 4) // 4:]), ) return opt @@ -1011,7 +1035,8 @@ def _read_opt_ssr(self, schema: 'Schema_SSROption', *, options: 'Option') -> 'Da type=self._read_ipv4_opt_type(schema.type), length=schema.length, pointer=schema.pointer, - route=tuple(schema.route), + route=tuple(schema.route[:(schema.pointer - 4) // 4]), + remaining=tuple(schema.route[(schema.pointer - 4) // 4:]), ) return opt @@ -1196,7 +1221,14 @@ def _read_opt_qs(self, schema: 'Schema_QSOption', *, options: 'Option') -> 'Data Raises: ProtocolError: If the option is malformed. + Note: + An option whose QS function is neither of the two above has no + defined layout, so the schema keeps it as an unassigned option and + it is parsed as one, c.f. :meth:`_read_opt_unassigned`. + """ + if isinstance(schema, Schema_UnassignedOption): + return self._read_opt_unassigned(schema, options=options) # type: ignore[return-value] if schema.length != 8: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format') @@ -1213,6 +1245,7 @@ def _read_opt_qs(self, schema: 'Schema_QSOption', *, options: 'Option') -> 'Data rate=40000 * (2 ** rate) / 1000 if rate > 0 else 0, ttl=datetime.timedelta(seconds=schema_req.ttl), nonce=schema_req.nonce['nonce'], + reserved=schema_req.nonce['reserved'], ) # type: Data_QSOption elif func == Enum_QSFunction.Report_of_Approved_Rate: schema_rep = cast('Schema_QuickStartReportOption', schema) @@ -1224,25 +1257,42 @@ def _read_opt_qs(self, schema: 'Schema_QSOption', *, options: 'Option') -> 'Data length=schema_rep.length, func=func, rate=40000 * (2 ** rate) / 1000 if rate > 0 else 0, + unused=schema_rep.reserved, nonce=schema_rep.nonce['nonce'], + reserved=schema_rep.nonce['reserved'], ) else: raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] unknown QS function: {func}') return opt - def _make_ipv4_options(self, options: 'list[Schema_Option | tuple[Enum_OptionNumber, dict[str, Any]] | bytes] | Option') -> 'tuple[list[Schema_Option | bytes], int]': + def _make_ipv4_options(self, options: 'list[Schema_Option | tuple[Enum_OptionNumber, dict[str, Any]] | bytes] | Option', + padding: 'Optional[bytes]' = None) -> 'tuple[list[Schema_Option | bytes], int]': """Make options for IPv4. + A :obj:`list` is a fresh option list, so any ``NOP`` and ``EOOL`` in it + are dropped and the padding is generated. A container is what + :meth:`read` hands back, so its ``NOP`` and ``EOOL`` options are written + in place, as parsed. + + Either way, the options are concatenated and then padded once, after the + last one, to the 32-bit boundary: with ``NOP`` options and a closing + ``EOOL``, or with zero octets if an ``EOOL`` has already ended the list. + ``padding`` goes after that ``EOOL`` verbatim, before any alignment; it + is dropped if no ``EOOL`` ends the list, since the reader would take its + octets for options. + Args: option: IPv4 options + padding: octets after the End of Option List Returns: Tuple of options and total length of options. """ total_length = 0 + terminated = False + options_list = [] # type: list[Schema_Option | bytes] if isinstance(options, list): - options_list = [] # type: list[Schema_Option | bytes] for schema in options: if isinstance(schema, bytes): code = Enum_OptionNumber.get(schema[0]) @@ -1276,69 +1326,62 @@ def _make_ipv4_options(self, options: 'list[Schema_Option | tuple[Enum_OptionNum options_list.append(data) total_length += data_len + else: + for code, option in options.items(multi=True): + name = self._lookup_registry(self.__option__, code) + if isinstance(name, str): + meth_name = f'_make_opt_{name}' + meth = cast('OptionConstructor', + getattr(self, meth_name, self._make_opt_unassigned)) + else: + meth = name[1] - # force alignment to 32-bit boundary - if data_len % 4: - pad_len = 4 - (data_len % 4) - # NOTE: The terminator is an EOOL option *schema*, like the NOP - # padding. The bare ``Enum_OptionNumber.EOOL`` wire code is not - # an option, and the ``options`` field takes only schemas and - # :obj:`bytes`, so packing would fail with ``FieldValueError: - # Field options has invalid value`` for any option whose - # length is not a multiple of four. - pad_opt = self._make_opt_nop(Enum_OptionNumber.NOP) # type: ignore[arg-type] - end_opt = self._make_opt_eool(Enum_OptionNumber.EOOL) # type: ignore[arg-type] - total_length += pad_len - - for _ in range(pad_len - 1): - options_list.append(pad_opt) - options_list.append(end_opt) - return options_list, total_length - - options_list = [] - for code, option in options.items(multi=True): - # ignore padding options by default - if code in (Enum_OptionNumber.NOP, Enum_OptionNumber.EOOL): - continue - - name = self._lookup_registry(self.__option__, code) - if isinstance(name, str): - meth_name = f'_make_opt_{name}' - meth = cast('OptionConstructor', - getattr(self, meth_name, self._make_opt_unassigned)) + data = meth(code, option) + data_len = len(data.pack()) + + options_list.append(data) + total_length += data_len + if code == Enum_OptionNumber.EOOL: + terminated = True + + # NOTE: Raw octets are safe here only after the EOOL: the reader + # stops there, and never looks at what follows as an option. + if terminated and padding: + options_list.append(bytes(padding)) + total_length += len(padding) + + # force alignment to 32-bit boundary + # + # NOTE: Once, after the last option. An EOOL after each unaligned + # option would end the list there, and every later option would be + # lost on parse. The terminator is an EOOL option *schema*, like the + # NOP padding: the ``options`` field takes only schemas and + # :obj:`bytes`, not the bare wire code. + if total_length % 4: + pad_len = 4 - (total_length % 4) + total_length += pad_len + + # zero octets after an EOOL already in the list, as EOOL schemas + if terminated: + pad_opt = cast('Schema_Option', self._make_opt_eool(Enum_OptionNumber.EOOL)) # type: ignore[arg-type] else: - meth = name[1] - - data = meth(code, option) - data_len = len(data.pack()) - - options_list.append(data) - total_length += data_len - - # force alignment to 32-bit boundary - if data_len % 4: - pad_len = 4 - (data_len % 4) - # NOTE: An EOOL option schema rather than the bare wire code, as in - # the list branch above. This is the ``from_data`` branch, since a - # parsed packet hands its options back as a container. - pad_opt = self._make_opt_nop(Enum_OptionNumber.NOP) # type: ignore[arg-type] - end_opt = self._make_opt_eool(Enum_OptionNumber.EOOL) # type: ignore[arg-type] - total_length += pad_len - - for _ in range(pad_len - 1): - options_list.append(pad_opt) - options_list.append(end_opt) + pad_opt = cast('Schema_Option', self._make_opt_nop(Enum_OptionNumber.NOP)) # type: ignore[arg-type] + end_opt = self._make_opt_eool(Enum_OptionNumber.EOOL) # type: ignore[arg-type] + + for _ in range(pad_len - 1): + options_list.append(pad_opt) + options_list.append(end_opt) return options_list, total_length def _make_opt_unassigned(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_UnassignedOption]' = None, *, - data: 'bytes', + data: 'bytes' = b'', **kwargs: 'Any') -> 'Schema_Option': """Make IPv4 unassigned options. Args: kind: option type code option: option data - data: option payload + data: option payload, i.e. the octets after the ``length`` octet **kwargs: arbitrary keyword arguments Returns: @@ -1350,7 +1393,7 @@ def _make_opt_unassigned(self, kind: 'Enum_OptionNumber', option: 'Optional[Data return Schema_UnassignedOption( type=kind, - length=len(data), + length=len(data) + 2, data=data, ) @@ -1418,10 +1461,10 @@ def _make_opt_sec(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SECOpt reserves as a field termination indicator. Notes: - :rfc:`1108` section 2.2 lays each protection authority octet out as + :rfc:`1108#section-2.4` lays each protection authority octet out as seven authority bits followed by a *field termination indicator* in - bit 0: ``0`` means another octet follows, ``1`` means this is the - last. So the authority numbering skips every position that is a + the low-order bit: ``1`` means another octet follows, ``0`` means + this is the last. So the authority numbering skips every position that is a termination bit -- 7, 15, 23 -- which is what :meth:`_read_opt_sec` encodes by looping over ``range(7)`` per octet, and the reason ``Field_Termination_Indicator`` is rejected @@ -1459,16 +1502,24 @@ def _make_opt_sec(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SECOpt # multiple of eight. max_auth = max(authorities) int_len = math.ceil((max_auth + 1) / 8) + else: + int_len = 0 + + # A parsed option keeps its own width, so that octets which carry no + # authority are not dropped on rebuild. + if option is not None: + int_len = max(int_len, option.length - 3) + if int_len: data_list = [b'0' for _ in range(int_len * 8)] for auth in authorities: data_list[auth] = b'1' - # Bit 0 of the *last* octet terminates the field. The intermediate - # octets keep their initial ``0``, meaning "another octet follows", - # so this one assignment is the whole indicator; without it the - # reader warns on every option written here. - data_list[-1] = b'1' + # The low-order bit of every octet but the last is ``1``, meaning + # "another octet follows"; the last keeps its ``0``, which ends the + # field. + for index in range(int_len - 1): + data_list[index * 8 + 7] = b'1' data = int(b''.join(data_list), base=2).to_bytes(int_len, 'big', signed=False) else: @@ -1484,6 +1535,7 @@ def _make_opt_sec(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SECOpt def _make_opt_lsr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_LSROption]' = None, *, counts: 'int' = 10, # reasonable default route: 'Optional[list[IPv4Address | str | bytes | int]]' = None, + remaining: 'Optional[list[IPv4Address | str | bytes | int]]' = None, **kwargs: 'Any') -> 'Schema_LSROption': """Make IPv4 Loose Source and Record Route (``LSR``) option. @@ -1491,7 +1543,10 @@ def _make_opt_lsr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_LSROpt kind: option type code option: option data counts: maximum number of addresses to record - route: list of IPv4 addresses as recorded routes + route: list of IPv4 addresses as recorded routes, i.e. the + address slots before the pointer + remaining: list of IPv4 addresses for the address slots at or beyond + the pointer; the slots left over up to ``counts`` are zero **kwargs: arbitrary keyword arguments Returns: @@ -1499,7 +1554,8 @@ def _make_opt_lsr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_LSROpt """ if option is not None: - route = cast('list[IPv4Address | str | bytes | int]', option.route) + slots = list(option.route or ()) # type: list[IPv4Address | str | bytes | int] + slots += getattr(option, 'remaining', ()) pointer = option.pointer length = option.length else: @@ -1507,11 +1563,16 @@ def _make_opt_lsr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_LSROpt length = 3 + counts * 4 pointer = 4 + min(len(route), counts) * 4 + # NOTE: Every address slot of the option is written, the empty ones + # as ``0.0.0.0``. + slots = (list(route[:counts]) + list(remaining or ()))[:counts] + slots += [0] * (counts - len(slots)) + return Schema_LSROption( type=kind, length=length, pointer=pointer, - route=route, + route=slots, ) def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOption]' = None, *, @@ -1536,22 +1597,22 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio ts_list = [] # type: list[int] if isinstance(option.timestamp, tuple): for ts in option.timestamp: + # NOTE: An integer is the raw 32-bit field, written as + # is: a set high-order bit marks a non-standard time value. if not isinstance(ts, int): ts = math.floor(ts.total_seconds() * 1000) - - if ts.bit_length() > 31: - warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning) - ts = ts | 0x80000000 + if ts.bit_length() > 31: + warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning) + ts = ts | 0x80000000 ts_list.append(ts) else: for ip, ts in option.timestamp.items(True): ts_list.append(int(ip)) if not isinstance(ts, int): ts = math.floor(ts.total_seconds() * 1000) - - if ts.bit_length() > 31: - warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning) - ts = ts | 0x80000000 + if ts.bit_length() > 31: + warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning) + ts = ts | 0x80000000 ts_list.append(ts) length = option.length @@ -1572,10 +1633,9 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio if not isinstance(ts, int): ts = math.floor(ts.total_seconds() * 1000) - - if ts.bit_length() > 31: - warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning) - ts = ts | 0x80000000 + if ts.bit_length() > 31: + warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning) + ts = ts | 0x80000000 ts_list.append(ts) elif isinstance(timestamp, dict): flag = Enum_TSFlag.IP_with_Timestamp # type: ignore[assignment] @@ -1590,12 +1650,11 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio ts_list.append(int(ip)) if not isinstance(ts, int): ts = math.floor(ts.total_seconds() * 1000) + if ts.bit_length() > 31: + warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning) + ts = ts | 0x80000000 if ts == 0: flag = Enum_TSFlag.Prespecified_IP_with_Timestamp # type: ignore[assignment] - - if ts.bit_length() > 31: - warn(f'{self.alias}: [OptNo {kind}] timestamp value is too large: {ts}', ProtocolWarning) - ts = ts | 0x80000000 ts_list.append(ts) else: raise ProtocolError(f'{self.alias}: [OptNo {kind}] invalid timestamp value: {timestamp}') @@ -1653,6 +1712,7 @@ def _make_opt_e_sec(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_ESEC def _make_opt_rr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_RROption]' = None, *, counts: 'int' = 10, # reasonable default route: 'Optional[list[IPv4Address | str | bytes | int]]' = None, + remaining: 'Optional[list[IPv4Address | str | bytes | int]]' = None, **kwargs: 'Any') -> 'Schema_RROption': """Make IPv4 Record Route (``RR``) option. @@ -1660,7 +1720,10 @@ def _make_opt_rr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_RROptio kind: option type code option: option data counts: maximum number of addresses to record - route: list of IPv4 addresses as recorded routes + route: list of IPv4 addresses as recorded routes, i.e. the + address slots before the pointer + remaining: list of IPv4 addresses for the address slots at or beyond + the pointer; the slots left over up to ``counts`` are zero **kwargs: arbitrary keyword arguments Returns: @@ -1668,7 +1731,8 @@ def _make_opt_rr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_RROptio """ if option is not None: - route = cast('list[IPv4Address | str | bytes | int]', option.route) + slots = list(option.route or ()) # type: list[IPv4Address | str | bytes | int] + slots += getattr(option, 'remaining', ()) pointer = option.pointer length = option.length else: @@ -1676,11 +1740,16 @@ def _make_opt_rr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_RROptio length = 3 + counts * 4 pointer = 4 + min(len(route), counts) * 4 + # NOTE: Every address slot of the option is written, the empty ones + # as ``0.0.0.0``. + slots = (list(route[:counts]) + list(remaining or ()))[:counts] + slots += [0] * (counts - len(slots)) + return Schema_RROption( type=kind, length=length, pointer=pointer, - route=route, + route=slots, ) def _make_opt_sid(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SIDOption]' = None, *, @@ -1710,6 +1779,7 @@ def _make_opt_sid(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SIDOpt def _make_opt_ssr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SSROption]' = None, *, counts: 'int' = 10, # reasonable default route: 'Optional[list[IPv4Address | str | bytes | int]]' = None, + remaining: 'Optional[list[IPv4Address | str | bytes | int]]' = None, **kwargs: 'Any') -> 'Schema_SSROption': """Make IPv4 Strict Source Route (``SSR``) option. @@ -1717,7 +1787,10 @@ def _make_opt_ssr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SSROpt kind: option type code option: option data counts: maximum number of addresses to record - route: list of IPv4 addresses as recorded routes + route: list of IPv4 addresses as recorded routes, i.e. the + address slots before the pointer + remaining: list of IPv4 addresses for the address slots at or beyond + the pointer; the slots left over up to ``counts`` are zero **kwargs: arbitrary keyword arguments Returns: @@ -1725,7 +1798,8 @@ def _make_opt_ssr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SSROpt """ if option is not None: - route = cast('list[IPv4Address | str | bytes | int]', option.route) + slots = list(option.route or ()) # type: list[IPv4Address | str | bytes | int] + slots += getattr(option, 'remaining', ()) pointer = option.pointer length = option.length else: @@ -1733,11 +1807,16 @@ def _make_opt_ssr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_SSROpt length = 3 + counts * 4 pointer = 4 + min(len(route), counts) * 4 + # NOTE: Every address slot of the option is written, the empty ones + # as ``0.0.0.0``. + slots = (list(route[:counts]) + list(remaining or ()))[:counts] + slots += [0] * (counts - len(slots)) + return Schema_SSROption( type=kind, length=length, pointer=pointer, - route=route, + route=slots, ) def _make_opt_mtup(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_MTUPOption]' = None, *, @@ -1857,15 +1936,17 @@ def _make_opt_rtralt(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_RTR alert=alert_val, ) - def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickStartRequestOption | Data_QuickStartReportOption]' = None, *, + def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickStartRequestOption | Data_QuickStartReportOption | Data_UnassignedOption]' = None, *, func: 'Enum_QSFunction | StdlibEnum | AenumEnum | str | int' = Enum_QSFunction.Quick_Start_Request, func_default: 'Optional[int]' = None, func_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long func_reversed: 'bool' = False, rate: 'int' = 0, ttl: 'timedelta | int' = 0, + unused: 'int' = 0, nonce: 'int' = 0, - **kwargs: 'Any') -> 'Schema_QSOption': + reserved: 'int' = 0, + **kwargs: 'Any') -> 'Schema_QSOption | Schema_UnassignedOption': """Make IPv4 Quick-Start (``QS``) option. Args: @@ -1877,26 +1958,37 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt func_reversed: reversed flag for QS function type rate: rate (in kbps) ttl: time to live (in seconds) + unused: the not-used fourth octet of a Report of Approved Rate nonce: nonce value + reserved: reserved bits (``R``) after the nonce **kwargs: arbitrary keyword arguments Returns: - Constructed option schema. + Constructed option schema. An unassigned option schema if ``option`` + is a Quick-Start option parsed as unassigned, c.f. :meth:`_read_opt_qs`, + or if ``func`` is neither of the assigned functions, in which case + the option takes the eight-octet layout of a Quick-Start Request. + + Raises: + ProtocolError: If ``func`` does not fit in four bits. """ + if isinstance(option, Data_UnassignedOption): + return cast('Schema_UnassignedOption', self._make_opt_unassigned(kind, option)) if option is not None: func_enum = option.func rate = option.rate ttl = getattr(option, 'ttl', 0) + unused = getattr(option, 'unused', 0) nonce = option.nonce + reserved = getattr(option, 'reserved', 0) else: func_enum = self._make_index(func, func_default, namespace=func_namespace, # type: ignore[assignment] reversed=func_reversed, pack=False) rate_val = math.floor(math.log2(rate * 1000 / 40000)) if rate > 0 else 0 + ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds()) if func_enum == Enum_QSFunction.Quick_Start_Request: - ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds()) - return Schema_QuickStartRequestOption( type=kind, length=8, @@ -1907,6 +1999,7 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt ttl=ttl_value, nonce={ 'nonce': nonce, + 'reserved': reserved, }, ) if func_enum == Enum_QSFunction.Report_of_Approved_Rate: @@ -1917,8 +2010,13 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt 'func': func_enum, 'rate': rate_val, }, + reserved=unused, nonce={ 'nonce': nonce, + 'reserved': reserved, }, ) - raise ProtocolError(f'{self.alias}: [OptNo {kind}] invalid QS function: {func_enum}') + if not 0 <= func_enum < 16: + raise ProtocolError(f'{self.alias}: [OptNo {kind}] invalid QS function: {func_enum}') + data = bytes([(func_enum << 4) | rate_val, ttl_value]) + ((nonce << 2) | reserved).to_bytes(4, 'big') + return cast('Schema_UnassignedOption', self._make_opt_unassigned(kind, data=data)) diff --git a/pcapkit/protocols/schema/internet/ipv4.py b/pcapkit/protocols/schema/internet/ipv4.py index 67da61353c..9e0cabfcee 100644 --- a/pcapkit/protocols/schema/internet/ipv4.py +++ b/pcapkit/protocols/schema/internet/ipv4.py @@ -20,7 +20,7 @@ from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField from pcapkit.corekit.multidict import OrderedMultiDict from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final -from pcapkit.utilities.exceptions import FieldValueError +from pcapkit.utilities.exceptions import EnumValueError, FieldValueError from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING from pcapkit.utilities.warnings import ProtocolWarning, warn @@ -67,6 +67,8 @@ class VerIHLField(TypedDict): class Flags(TypedDict): """Flags and fragment offset field.""" + #: Reserved flag bit. + reserved: int #: Don't fragment flag. df: int #: More fragments flag. @@ -93,6 +95,8 @@ class QuickStartFlags(TypedDict): class QSTestFlags(TypedDict): """Quick start test flag.""" + #: Option length. + length: int #: QS function. func: int @@ -101,6 +105,8 @@ class QSNonce(TypedDict): #: Nonce. nonce: int + #: Reserved bits. + reserved: int def quick_start_option_length(schema: 'Type[QSOption]') -> 'int': @@ -194,6 +200,9 @@ def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field': * If ``func`` is ``8``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` wrapped :class:`~pcapkit.protocols.schema.internet.ipv4.QuickStartReportOption` instance. + * Otherwise, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField` + wrapped :class:`~pcapkit.protocols.schema.internet.ipv4.UnassignedOption` + instance, sized by the option's own ``length`` octet. Notes: The length handed to the :class:`~pcapkit.corekit.fields.misc.SchemaField` @@ -203,12 +212,21 @@ def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field': leftover octets are read as a further, fabricated option. """ - func = Enum_QSFunction.get(pkt['flags']['func']) - pkt['flags']['func'] = func + # NOTE: The enumeration admits ``0`` to ``8`` only, while the field is four + # bits wide; a function above ``8`` is just as unassigned. + try: + func = Enum_QSFunction.get(pkt['flags']['func']) + except EnumValueError: + schema = None + else: + pkt['flags']['func'] = func + schema = QSOption.registry[func] - schema = QSOption.registry[func] if schema is None: - raise FieldValueError(f'IPv4: invalid QS function: {func}') + # An unassigned function has no defined layout, so the option is kept + # as raw data, sized by its own ``length`` octet: two octets at least, + # so that a shorter length reaches the reader and is rejected there. + return SchemaField(length=max(pkt['flags']['length'], 2), schema=UnassignedOption) return SchemaField(length=quick_start_option_length(schema), schema=schema) @@ -290,15 +308,15 @@ class LSROption(Option, code=Enum_OptionNumber.LSR): #: Pointer. pointer: 'int' = UInt8Field() - #: Route. + #: Route: every address slot of the option, those before the pointer and + #: those at or beyond it alike. route: 'list[IPv4Address]' = ListField( - length=lambda pkt: pkt['pointer'] - 4, + length=lambda pkt: max(pkt['length'] - 3, 0) // 4 * 4, item_type=IPv4AddressField(), ) - #: Remaining data buffer. + #: Trailing octets that do not make up a whole address slot. remainder: 'bytes' = PaddingField( - length=lambda pkt: pkt['length'] - pkt['pointer'] + 1, - default=bytes(36), # a reasonable default + length=lambda pkt: max(pkt['length'] - 3, 0) % 4, ) if TYPE_CHECKING: @@ -369,9 +387,10 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': for ts in ts_data: self.data.append(ts) + # NOTE: A set high-order bit marks a non-standard time value, + # per :rfc:`791`, so it is kept as the raw 32-bit integer. if ts >> 31: - warn(f'IPv4: [OptNo {self.type}] invalid format: timestamp error: {ts}', ProtocolWarning) - ts_val = ts & 0x7FFFFFFF # type: int | timedelta + ts_val = ts # type: int | timedelta else: ts_val = datetime.timedelta(milliseconds=ts) ts_list.append(ts_val) @@ -387,8 +406,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': self.data.add(ip_val, ts) if ts >> 31: - warn(f'IPv4: [OptNo {self.type}] invalid format: timestamp error: {ts}', ProtocolWarning) - ts_val = ts & 0x7FFFFFFF + ts_val = ts else: ts_val = datetime.timedelta(milliseconds=ts) timestamp.add(ip_val, ts_val) @@ -403,8 +421,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': self.data.add(ip_val, ts) if ts >> 31: - warn(f'IPv4: [OptNo {self.type}] invalid format: timestamp error: {ts}', ProtocolWarning) - ts_val = ts & 0x7FFFFFFF + ts_val = ts else: ts_val = datetime.timedelta(milliseconds=ts) timestamp.add(ip_val, ts_val) @@ -476,15 +493,15 @@ class RROption(Option, code=Enum_OptionNumber.RR): #: Pointer. pointer: 'int' = UInt8Field() - #: Route. + #: Route: every address slot of the option, those before the pointer and + #: those at or beyond it alike. route: 'list[IPv4Address]' = ListField( - length=lambda pkt: pkt['pointer'] - 4, + length=lambda pkt: max(pkt['length'] - 3, 0) // 4 * 4, item_type=IPv4AddressField(), ) - #: Remaining data buffer. + #: Trailing octets that do not make up a whole address slot. remainder: 'bytes' = PaddingField( - length=lambda pkt: pkt['length'] - pkt['pointer'] + 1, - default=bytes(36), # a reasonable default + length=lambda pkt: max(pkt['length'] - 3, 0) % 4, ) if TYPE_CHECKING: @@ -510,15 +527,15 @@ class SSROption(Option, code=Enum_OptionNumber.SSR): #: Pointer. pointer: 'int' = UInt8Field() - #: Route. + #: Route: every address slot of the option, those before the pointer and + #: those at or beyond it alike. route: 'list[IPv4Address]' = ListField( - length=lambda pkt: pkt['pointer'] - 4, + length=lambda pkt: max(pkt['length'] - 3, 0) // 4 * 4, item_type=IPv4AddressField(), ) - #: Remaining data buffer. + #: Trailing octets that do not make up a whole address slot. remainder: 'bytes' = PaddingField( - length=lambda pkt: pkt['length'] - pkt['pointer'] + 1, - default=bytes(36), # a reasonable default + length=lambda pkt: max(pkt['length'] - 3, 0) % 4, ) if TYPE_CHECKING: @@ -581,6 +598,7 @@ class _QSOption(Schema): #: Flags. flags: 'QSTestFlags' = ForwardMatchField(BitField(length=3, namespace={ + 'length': (8, 8), 'func': (16, 4), })) #: QS data. @@ -588,7 +606,7 @@ class _QSOption(Schema): selector=quick_start_data_selector, ) - def post_process(self, packet: 'dict[str, Any]') -> 'QSOption': + def post_process(self, packet: 'dict[str, Any]') -> 'QSOption | UnassignedOption': """Revise ``schema`` data after unpacking process. Args: @@ -599,7 +617,8 @@ def post_process(self, packet: 'dict[str, Any]') -> 'QSOption': """ ret = self.data - ret.func = Enum_QSFunction.get(self.flags['func']) + if isinstance(ret, QSOption): + ret.func = Enum_QSFunction.get(self.flags['func']) return ret @@ -631,6 +650,7 @@ class QuickStartRequestOption(QSOption, code=Enum_QSFunction.Quick_Start_Request #: QS nonce. nonce: 'QSNonce' = BitField(length=4, namespace={ 'nonce': (0, 30), + 'reserved': (30, 2), }) if TYPE_CHECKING: @@ -653,18 +673,18 @@ class QuickStartReportOption(QSOption, code=Enum_QSFunction.Report_of_Approved_R #: :meth:`~pcapkit.protocols.internet.ipv4.IPv4._make_opt_qs` writes into #: ``length`` and the eight #: :meth:`~pcapkit.protocols.internet.ipv4.IPv4._read_opt_qs` demands of it, - #: and the ``nonce`` would decode one octet early. Declared as padding rather - #: than as data because :rfc:`4782` gives it no meaning and no caller should - #: be setting it. - reserved: 'bytes' = PaddingField(length=1) + #: and the ``nonce`` would decode one octet early. Kept as an integer rather + #: than as padding, so that whatever the sender put there is written back. + reserved: 'int' = UInt8Field() #: QS nonce. nonce: 'QSNonce' = BitField(length=4, namespace={ 'nonce': (0, 30), + 'reserved': (30, 2), }) if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionNumber', length: 'int', flags: 'QuickStartFlags', - nonce: 'QSNonce') -> 'None': ... + reserved: 'int', nonce: 'QSNonce') -> 'None': ... @schema_final @@ -690,6 +710,7 @@ class IPv4(Schema): id: 'int' = UInt16Field() #: Flags and fragment offset. flags: 'Flags' = BitField(length=2, namespace={ + 'reserved': (0, 1), 'df': (1, 1), 'mf': (2, 1), 'offset': (3, 13), diff --git a/tests/protocols/internet/test_ipv4_roundtrip_unit.py b/tests/protocols/internet/test_ipv4_roundtrip_unit.py new file mode 100644 index 0000000000..130dc9650f --- /dev/null +++ b/tests/protocols/internet/test_ipv4_roundtrip_unit.py @@ -0,0 +1,215 @@ +# -*- coding: utf-8 -*- +"""IPv4 headers rebuild byte for byte, and ``make()`` keeps every option. + +Pins the fixes for GitHub issues #1316 to #1322 and #1326, found by the +internet-v4 round-trip audit (#1202): + +* #1316 -- ``make()`` padded with an EOOL after every unaligned option, so the + reader stopped there and the later options were lost; +* #1317 -- ``from_data`` dropped the parsed NOP/EOOL layout and the octets after + the EOOL, and re-padded per option. The octets after the EOOL are now the + ``padding`` data field, written back verbatim; the header length follows + from the options and those octets, so edited options recompute it; +* #1318 -- an option with an unassigned code could be neither rebuilt nor built; +* #1319 -- LSR, SSR and RR zeroed the route slots at and beyond the pointer; +* #1320 -- TS dropped the high-order "non-standard" bit of a timestamp; +* #1321 -- the SEC field termination indicator was the inverse of + :rfc:`1108` section 2.4(a), where the low-order bit is ``0`` on the final + octet and ``1`` when more follow; +* #1322 -- an unassigned Quick-Start function failed the whole header; +* #1326 -- the reserved flag bit and the Quick-Start reserved bits and unused + octet were written back as zero. + +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 +import warnings + +from tests._support import reimport_once_per_class + + +def header(options: 'str' = '', flags: 'str' = '0000') -> 'bytes': + """An IPv4 header from 10.0.0.1 to 10.0.0.2, protocol 253, with ``options`` in hex.""" + opts = bytes.fromhex(options) + ihl = 5 + len(opts) // 4 + return (bytes([0x40 | ihl, 0]) + (ihl * 4).to_bytes(2, 'big') + bytes(2) + + bytes.fromhex(flags) + bytes.fromhex('40fd0000') + + bytes([10, 0, 0, 1, 10, 0, 0, 2]) + opts) + + +#: Wire headers that must rebuild through ``from_data`` unchanged, by issue. +HEADERS = { + '1317 NOP then RR': header('0107070400000000'), + '1317 RR, RTRALT, EOOL and padding': header('070704000000009404000000'), + '1317 NOPs, RTRALT, EOOL and padding': header('010101940400000000000000'), + '1317 NOP x4': header('01010101'), + '1317 EOOL x4': header('00000000'), + '1317 EOOL, non-zero octets': header('00ff0102'), + '1317 EOOL, RR-like octets': header('00070304'), + '1317 RTRALT, EOOL, non-zero octets': header('9404000000abcdef'), + '1318 unassigned option': header('9e04abcd'), + '1318 unassigned option, no data': header('9e020000'), + '1319 LSR, both slots past the pointer': header('830b040a0000010a00000200'), + '1319 SSR, one slot past the pointer': header('890b080a0000010a00000200'), + '1319 RR, one slot past the pointer': header('070b080a0000010a00000200'), + '1320 TS, non-standard timestamp': header('4408090080000001'), + '1321 SEC, one authority octet': header('8204ab90'), + '1321 SEC, two authority octets': header('8205ab9100' '000000'), + '1322 QS, unassigned function 3': header('1908310500000000'), + '1322 QS, unassigned function 15': header('1908f00000000000'), + '1326 reserved flag bit': header(flags='8000'), + '1326 QS request, reserved bits': header('1908030500000003'), + '1326 QS report, unused octet': header('1908810500000000'), + '1326 QS report, reserved bits': header('1908810000000002'), +} + + +class TestIPv4RoundTrip(unittest.TestCase): + """Pin the IPv4 round trip and the ``make()`` option layout.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def parse(self, data: 'bytes') -> 'object': + from pcapkit.protocols.internet.ipv4 import IPv4 + + return IPv4(io.BytesIO(data), len(data)) + + def test_from_data_rebuilds_byte_for_byte(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + from pcapkit.utilities.warnings import ProtocolWarning + + for name, data in HEADERS.items(): + with self.subTest(case=name): + with warnings.catch_warnings(): + warnings.simplefilter('error', ProtocolWarning) + parsed = self.parse(data) + self.assertEqual(IPv4.from_data(parsed.info).data, data) + + def test_make_pads_once_after_the_last_option(self) -> None: + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + + cases = { + 'RR twice': ([(OptionNumber.RR, {'counts': 1}), (OptionNumber.RR, {'counts': 1})], + '0707040000000007070400000000' '0100', + [OptionNumber.RR, OptionNumber.RR, OptionNumber.NOP, OptionNumber.EOOL]), + 'RR then RTRALT': ([(OptionNumber.RR, {'counts': 1}), (OptionNumber.RTRALT, {})], + '07070400000000' '94040001' '00', + [OptionNumber.RR, OptionNumber.RTRALT, OptionNumber.EOOL]), + } + for name, (options, wire, codes) in cases.items(): + with self.subTest(case=name): + data = IPv4(options=options, protocol=253).data + self.assertEqual(data[20:].hex(), wire) + parsed = self.parse(data) + self.assertEqual([code for code, _ in parsed.info.options.items(multi=True)], codes) + + def test_make_builds_an_unassigned_option(self) -> None: + from pcapkit.protocols.internet.ipv4 import IPv4 + + data = IPv4(options=[(0x9e, {'data': b'\xab\xcd'})], protocol=253).data + self.assertEqual(data[20:].hex(), '9e04abcd') + self.assertEqual(self.parse(data).info.options[0x9e].data, b'\xab\xcd') + + def test_make_builds_an_unassigned_quick_start_function(self) -> None: + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.data.internet.ipv4 import UnassignedOption + from pcapkit.protocols.internet.ipv4 import IPv4 + + data = IPv4(options=[(OptionNumber.QS, {'func': 15, 'ttl': 1, 'nonce': 2})], + protocol=253).data + self.assertEqual(data[20:].hex(), '1908f00100000008') + option = self.parse(data).info.options[OptionNumber.QS] + self.assertIsInstance(option, UnassignedOption) + self.assertEqual(option.data, bytes.fromhex('f00100000008')) + + def test_route_slots_past_the_pointer_are_data(self) -> None: + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + + option = self.parse(HEADERS['1319 SSR, one slot past the pointer']).info.options[OptionNumber.SSR] + self.assertEqual([str(ip) for ip in option.route], ['10.0.0.1']) + self.assertEqual([str(ip) for ip in option.remaining], ['10.0.0.2']) + + # A fresh option writes the pending hops after the recorded ones, and + # zero-fills the slots left over up to ``counts``. + data = IPv4(options=[(OptionNumber.LSR, {'counts': 3, 'route': ['10.0.0.1'], + 'remaining': ['10.0.0.2']})], + protocol=253).data + self.assertEqual(data[20:].hex(), '830f08' '0a000001' '0a000002' '00000000' '00') + + def test_sec_termination_indicator_follows_rfc_1108(self) -> None: + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.const.ipv4.protection_authority import ProtectionAuthority + from pcapkit.utilities.warnings import ProtocolWarning + + with warnings.catch_warnings(): + warnings.simplefilter('error', ProtocolWarning) + option = self.parse(header('8204ab90')).info.options[OptionNumber.SEC] + self.assertEqual(option.flags, (ProtectionAuthority.GENSER, ProtectionAuthority.NSA)) + + # The low-order bit set on the final octet promises another octet. + with self.assertWarnsRegex(ProtocolWarning, 'field termination indicator not set'): + self.parse(header('8204ab91')) + + def test_timestamp_keeps_the_high_order_bit(self) -> None: + from pcapkit.const.ipv4.option_number import OptionNumber + + option = self.parse(HEADERS['1320 TS, non-standard timestamp']).info.options[OptionNumber.TS] + self.assertEqual(option.timestamp, (0x80000001,)) + + def test_reserved_bits_are_data(self) -> None: + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + + self.assertEqual(self.parse(HEADERS['1326 reserved flag bit']).info.flags.reserved, 1) + self.assertEqual(IPv4(reserved=1, protocol=253).data[6:8].hex(), '8000') + + request = self.parse(HEADERS['1326 QS request, reserved bits']).info.options[OptionNumber.QS] + self.assertEqual(request.reserved, 3) + report = self.parse(HEADERS['1326 QS report, unused octet']).info.options[OptionNumber.QS] + self.assertEqual(report.unused, 5) + + def test_octets_after_eool_are_data(self) -> None: + for name, octets in (('1317 EOOL, non-zero octets', 'ff0102'), + ('1317 RTRALT, EOOL, non-zero octets', 'abcdef'), + ('1317 NOP x4', '')): + with self.subTest(case=name): + self.assertEqual(self.parse(HEADERS[name]).info.padding.hex(), octets) + + def test_edited_options_recompute_the_header_length(self) -> None: + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + + rtralt = self.parse(header('94040000')).info.options[OptionNumber.RTRALT] + + # one option added to a parsed header grows it + parsed = self.parse(header('88040001')) + parsed.info.options.add(OptionNumber.RTRALT, rtralt) + data = IPv4.from_data(parsed.info).data + self.assertEqual(data[0], 0x47) + self.assertEqual(data[20:].hex(), '88040001' '94040000') + + # one option removed shrinks it, rather than zero-filling the old length + parsed = self.parse(header('88040001' '94040000')) + parsed.info.options.pop(OptionNumber.RTRALT) + data = IPv4.from_data(parsed.info).data + self.assertEqual(data[0], 0x46) + self.assertEqual(data[20:].hex(), '88040001') + + def test_padding_needs_an_eool_before_it(self) -> None: + from pcapkit.const.ipv4.option_number import OptionNumber + from pcapkit.protocols.internet.ipv4 import IPv4 + + # A fresh list drops its EOOL and pads once, so the octets go nowhere. + data = IPv4(options=[(OptionNumber.RR, {'counts': 1})], padding=b'\xff', protocol=253).data + self.assertEqual(data[20:].hex(), '07070400000000' '00') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_ipv4_unit.py b/tests/protocols/internet/test_ipv4_unit.py index b84481a137..b794912fe1 100644 --- a/tests/protocols/internet/test_ipv4_unit.py +++ b/tests/protocols/internet/test_ipv4_unit.py @@ -38,8 +38,9 @@ class DummyDict(dict): data = DummyDict( tos=DummyDict(pre=0, thr=False, rel=False, ecn=0, **{'del': False}), len=1186, + padding=b'', id=7, - flags=DummyDict(df=True, mf=False), + flags=DummyDict(reserved=0, df=True, mf=False), offset=0, ttl=timedelta(seconds=64), protocol=TransType.TCP, @@ -279,7 +280,7 @@ def header(value: 'object') -> 'Schema_IPv4': vihl={'version': 4, 'ihl': 5}, tos={'pre': 0, 'del': 0, 'thr': 0, 'rel': 0, 'ecn': 0}, length=20 + len(payload), id=0, - flags={'df': 0, 'mf': 0, 'offset': 0}, + flags={'reserved': 0, 'df': 0, 'mf': 0, 'offset': 0}, ttl=0, proto=6, chksum=b'\x00\x00', src='127.0.0.1', dst='127.0.0.2', options=[], payload=value, @@ -496,28 +497,21 @@ def test_ipv4_make_options_pads_with_an_eool_option_not_its_wire_code(self) -> N self.assertEqual(bytes(rebuilt), raw) def test_ipv4_make_opt_sec_sets_the_field_termination_indicator(self) -> None: - """A SEC option this library writes is one this library can read. C.f. #537. - - :rfc:`1108` section 2.2 makes bit 0 of each protection authority octet a - *field termination indicator*: ``0`` means another octet follows, ``1`` - means this is the last. ``_read_opt_sec`` enforces it, warning - ``field termination indicator not set`` when the final octet has it - clear. ``_make_opt_sec`` built the bitmap purely out of authority bit - positions and never set it, so *every* SEC option the library wrote with - at least one authority was one its own reader flagged as malformed -- - visible in the project's own generated capture - :file:`examples/captures/options-ipv4.pcap`, which warned - ``IPv4: [OptNo 130] invalid format: field termination indicator not set`` - on extraction. + """A SEC option this library writes is one this library can read. C.f. #537, #1321. + + :rfc:`1108` section 2.4(a) makes the low-order bit of each protection + authority octet a *field termination indicator*: ``1`` means another + octet follows, ``0`` means this is the last. ``_read_opt_sec`` enforces + it, warning ``field termination indicator not set`` when the final octet + has it set. #537 made the writer and the reader agree with each other on + the opposite convention; #1321 made both agree with the RFC. The round trip is asserted through a real datagram and with warnings - promoted to errors, because the defect's only symptom was a warning: - asserting on the octets alone would have let it back in, and asserting - that the flags survive alone would too -- they always did. + promoted to errors, because a disagreement's only symptom is a warning. - ``0x91`` rather than ``0x90`` is the whole change on the wire: - ``1001 0001``, bits 0 and 3 for ``GENSER`` and ``NSA`` counted from the - most significant, and bit 0 of the octet for the terminator. + ``0x90`` is ``1001 0000``: bits 0 and 3 for ``GENSER`` and ``NSA`` + counted from the most significant, and a clear low-order bit for the + final octet. """ from pcapkit.const.ipv4.option_number import OptionNumber @@ -529,15 +523,10 @@ def test_ipv4_make_opt_sec_sets_the_field_termination_indicator(self) -> None: authorities = [ProtectionAuthority.GENSER, ProtectionAuthority.NSA] schema = proto._make_opt_sec(OptionNumber.SEC, authorities=authorities) - self.assertEqual(schema.data, b'\x91') - self.assertEqual(schema.data[-1] & 0x01, 1) + self.assertEqual(schema.data, b'\x90') + self.assertEqual(schema.data[-1] & 0x01, 0) self.assertEqual(schema.length, 4) - # The reader's own predicate, stated here rather than inferred from the - # absence of a warning below, so a reader that stopped checking does not - # silently make this test vacuous. - self.assertNotEqual(schema.data[-1] & 0x01, 0) - # And the end-to-end statement: a datagram carrying this option parses # with no ProtocolWarning at all. The option is four octets -- type, # length, classification level and the one-octet bitmap -- which is @@ -583,32 +572,32 @@ def test_ipv4_make_opt_sec_sizes_the_bitmap_from_the_bit_count(self) -> None: proto = object.__new__(IPv4) - # One octet: GENSER at bit 0 from the most significant, and the + # One octet: GENSER at bit 0 from the most significant, and a clear # terminator. Formerly an IndexError. schema = proto._make_opt_sec(OptionNumber.SEC, authorities=[ProtectionAuthority.GENSER]) - self.assertEqual(schema.data, b'\x81') + self.assertEqual(schema.data, b'\x80') self.assertEqual(schema.length, 4) # The largest authority that still fits one octet, bit 6, since bit 7 is # the terminator. schema = proto._make_opt_sec(OptionNumber.SEC, authorities=[ProtectionAuthority(6)]) - self.assertEqual(schema.data, b'\x03') + self.assertEqual(schema.data, b'\x02') self.assertEqual(schema.length, 4) # Index 8 is the first bit of the *second* octet, so it needs two -- - # ceil(8/8) said one. The first octet is all zeros, its own terminator - # included, which is what says "another octet follows". + # ceil(8/8) said one. The first octet carries no authority, only its + # terminator, set to say "another octet follows". schema = proto._make_opt_sec(OptionNumber.SEC, authorities=[ProtectionAuthority(8)]) - self.assertEqual(schema.data, b'\x00\x81') + self.assertEqual(schema.data, b'\x01\x80') self.assertEqual(schema.length, 5) # Only the last octet terminates the field; an intermediate one that did # would make the reader warn 'remaining data'. - self.assertEqual(schema.data[0] & 0x01, 0) - self.assertEqual(schema.data[-1] & 0x01, 1) + self.assertEqual(schema.data[0] & 0x01, 1) + self.assertEqual(schema.data[-1] & 0x01, 0) # No authorities at all stays a bare 3-octet option with no bitmap, so # there is no final octet to terminate. @@ -622,8 +611,7 @@ def test_ipv4_make_opt_sec_rejects_a_termination_bit_as_an_authority(self) -> No ``Enum_ProtectionAuthority`` member ``7`` is named ``Field_Termination_Indicator`` -- it is not an authority at all, yet it was a member of the enumeration the writer accepted as one. Passing it - produced ``data=b'\\x01'``: an option that reads as validly terminated - while encoding zero authorities. + produced an option encoding zero authorities. The writer and the reader disagreed about whether index 7 is data, and this test records which of the two won. ``_read_opt_sec`` loops over @@ -682,7 +670,7 @@ def test_ipv4_make_opt_sec_rejects_a_termination_bit_as_an_authority(self) -> No self.assertEqual( proto._make_opt_sec(OptionNumber.SEC, authorities=[ProtectionAuthority.DOE]).data, - b'\x09', + b'\x08', ) def test_ipv4_sid_option_is_four_octets_wide_on_the_wire(self) -> None: @@ -1019,7 +1007,7 @@ def test_ipv4_properties_read_and_make_cover_packet_paths(self) -> None: tos={'pre': 0, 'del': 0, 'thr': 0, 'rel': 0, 'ecn': 0}, length=32, id=7, - flags={'df': 1, 'mf': 0, 'offset': 1}, + flags={'reserved': 0, 'df': 1, 'mf': 0, 'offset': 1}, ttl=64, proto=TransType.TCP, chksum=b'\x12\x34', @@ -1048,7 +1036,7 @@ def test_ipv4_properties_read_and_make_cover_packet_paths(self) -> None: tos={'pre': 0, 'del': 0, 'thr': 0, 'rel': 0, 'ecn': 0}, length=24, id=8, - flags={'df': 0, 'mf': 0, 'offset': 0}, + flags={'reserved': 0, 'df': 0, 'mf': 0, 'offset': 0}, ttl=32, proto=TransType.UDP, chksum=b'\x00\x00', @@ -1067,7 +1055,7 @@ def test_ipv4_properties_read_and_make_cover_packet_paths(self) -> None: tos={'pre': 0, 'del': 0, 'thr': 0, 'rel': 0, 'ecn': 0}, length=20, id=1, - flags={'df': 0, 'mf': 0, 'offset': 0}, + flags={'reserved': 0, 'df': 0, 'mf': 0, 'offset': 0}, ttl=1, proto=TransType.UDP, chksum=b'\x00\x00', @@ -1278,19 +1266,11 @@ def test_ipv4_unregistered_option_code_does_not_mutate_the_registry(self) -> Non proto._make_ipv4_options([(code, {'data': b'xx'})]) self.assertEqual(set(registry), before) - # The OrderedMultiDict branch cannot get as far as building the - # option: :meth:`IPv4._make_opt_unassigned - # ` - # declares ``data`` keyword-only with no default, where every - # sibling fallback constructor defaults it to ``b''``, so it - # raises before reading the payload out of ``option.data``. That - # predates the registry migration and is left alone here; what - # this asserts is that the *lookup* which runs first still does - # not insert. - with self.assertRaises(TypeError): - proto._make_ipv4_options(OrderedMultiDict([ - (code, types.SimpleNamespace(data=b'xx')), - ])) + # The OrderedMultiDict branch too, which builds the option out + # of ``option.data`` since #1318 gave ``data`` a default. + proto._make_ipv4_options(OrderedMultiDict([ + (code, types.SimpleNamespace(data=b'xx')), + ])) self.assertEqual(set(registry), before) # a leak would make the next genuine registration warn @@ -1311,25 +1291,22 @@ def test_ipv4_option_constructors_cover_common_and_error_branches(self) -> None: proto = object.__new__(IPv4) + # ``length`` counts the type and length octets too. C.f. #1318. unknown = proto._make_opt_unassigned(OptionNumber.get(31), data=b'abc') - self.assertEqual(unknown.to_dict()['length'], 3) + self.assertEqual(unknown.to_dict()['length'], 5) self.assertEqual(unknown.to_dict()['data'], b'abc') self.assertEqual(proto._make_opt_eool(OptionNumber.EOOL).to_dict()['length'], 1) self.assertEqual(proto._make_opt_nop(OptionNumber.NOP).to_dict()['length'], 1) - # ``GENSER`` alone, where this passed - # ``[GENSER, Field_Termination_Indicator]`` before #537: the expected - # octets are unchanged, because bit 0 of the last octet is now set by - # ``_make_opt_sec`` itself rather than by naming the terminator as though - # it were an authority -- which is what produced the 0x01 here, and what - # is now rejected. + # ``GENSER`` alone: the low-order bit of the last octet is clear, per + # :rfc:`1108` section 2.4(a). C.f. #537 and #1321. sec = proto._make_opt_sec( OptionNumber.SEC, authorities=[ProtectionAuthority.GENSER], ) self.assertEqual(sec.to_dict()['length'], 4) - self.assertEqual(sec.to_dict()['data'], b'\x81') + self.assertEqual(sec.to_dict()['data'], b'\x80') loose = proto._make_opt_lsr(OptionNumber.LSR, counts=2, route=['192.0.2.1']) self.assertEqual(loose.to_dict()['length'], 11) @@ -1388,11 +1365,9 @@ def test_ipv4_option_constructors_cover_common_and_error_branches(self) -> None: # Which is why the alignment branch gets a case of its own rather than # riding along on SID's old width. The padding terminator is an EOOL - # *option*, not the bare wire code, and these assertions pinned the - # latter -- that is what #506 fixed, and it would have been lost here - # when SID stopped being a trigger. A SEC option carrying a two-octet - # bitmap packs to five, so it still is one: three octets of padding, two - # NOPs and the EOOL. + # *option*, not the bare wire code -- that is what #506 fixed. A SEC + # option carrying a two-octet bitmap packs to five, so it is one: three + # octets of padding, two NOPs and the EOOL. padded, padded_length = proto._make_ipv4_options([ (OptionNumber.SEC, {'authorities': [ProtectionAuthority(8)]}), ]) @@ -1455,6 +1430,7 @@ def opt_type(code): length=7, pointer=8, route=[ip_address('192.0.2.1')], + remaining=(), ) rr = ipv4_data.RROption( code=OptionNumber.RR, @@ -1462,6 +1438,7 @@ def opt_type(code): length=11, pointer=12, route=[ip_address('192.0.2.1'), ip_address('192.0.2.2')], + remaining=(), ) ssr = ipv4_data.SSROption( code=OptionNumber.SSR, @@ -1469,6 +1446,7 @@ def opt_type(code): length=7, pointer=8, route=[ip_address('192.0.2.3')], + remaining=(), ) self.assertEqual(proto._make_opt_lsr(OptionNumber.LSR, lsr).pointer, 8) self.assertEqual(proto._make_opt_rr(OptionNumber.RR, rr).pointer, 12) @@ -1483,10 +1461,13 @@ def opt_type(code): flag=TSFlag.Timestamp_Only, timestamp=(datetime.timedelta(seconds=1), 0x80000000), ) + # An integer is the raw field, its high-order bit marking a non-standard + # value, so it is written as is and without a warning. C.f. #1320. with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: ts_schema = proto._make_opt_ts(OptionNumber.TS, ts_tuple) self.assertEqual(ts_schema.flags['oflw'], 1) - warn.assert_called_once() + self.assertEqual(ts_schema.ts_data[1], 0x80000000) + warn.assert_not_called() ts_map = OrderedMultiDict([ (ip_address('192.0.2.10'), datetime.timedelta(seconds=2)), @@ -1504,13 +1485,18 @@ def opt_type(code): with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: self.assertEqual(proto._make_opt_ts(OptionNumber.TS, ts_prespecified).flags['flag'], TSFlag.Prespecified_IP_with_Timestamp) - warn.assert_called_once() + warn.assert_not_called() with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: proto._make_opt_ts(OptionNumber.TS, counts=1, timestamp=[1, 2]) warn.assert_called_once() with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: proto._make_opt_ts(OptionNumber.TS, counts=1, timestamp=[0x80000000]) + warn.assert_not_called() + # A timedelta too large for 31 bits still warns. + with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: + proto._make_opt_ts(OptionNumber.TS, counts=1, + timestamp=[datetime.timedelta(milliseconds=0x80000000)]) warn.assert_called_once() with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: proto._make_opt_ts(OptionNumber.TS, counts=1, timestamp={ @@ -1521,7 +1507,7 @@ def opt_type(code): with mock.patch('pcapkit.protocols.internet.ipv4.warn') as warn: proto._make_opt_ts(OptionNumber.TS, counts=1, timestamp={ip_address('192.0.2.22'): 0x80000000}) - warn.assert_called_once() + warn.assert_not_called() e_sec = ipv4_data.ESECOption( code=OptionNumber.E_SEC, @@ -1579,17 +1565,22 @@ def opt_type(code): rate=80, ttl=datetime.timedelta(seconds=7), nonce=3, + reserved=2, ) self.assertEqual(proto._make_opt_qs(OptionNumber.QS, qs_request).ttl, 7) + self.assertEqual(proto._make_opt_qs(OptionNumber.QS, qs_request).nonce['reserved'], 2) qs_report = ipv4_data.QuickStartReportOption( code=OptionNumber.QS, type=opt_type(OptionNumber.QS), length=8, func=QSFunction.Report_of_Approved_Rate, rate=80, + unused=5, nonce=3, + reserved=1, ) self.assertEqual(proto._make_opt_qs(OptionNumber.QS, qs_report).nonce['nonce'], 3) + self.assertEqual(proto._make_opt_qs(OptionNumber.QS, qs_report).reserved, 5) schema_options, schema_total = proto._make_ipv4_options([ bytes([OptionNumber.NOP]), @@ -1614,11 +1605,7 @@ def opt_type(code): (OptionNumber.SID, sid), (OptionNumber.LSR, lsr), # A SEC option whose authorities reach into a second bitmap octet, so - # it packs to five and needs three octets of padding -- two NOPs and - # the EOOL. Which is what keeps the multi-NOP arm of the alignment - # branch exercised on this, the data-model path: SID used to reach it - # by being two octets over-wide (6 % 4 == 2, one NOP), and since #534 - # it is 4-aligned and reaches it not at all. + # it packs to five. (OptionNumber.SEC, ipv4_data.SECOption( code=OptionNumber.SEC, type=opt_type(OptionNumber.SEC), @@ -1629,20 +1616,19 @@ def opt_type(code): (OptionNumber.MTUP, mtup), ]) mapped_options, mapped_total = proto._make_ipv4_options(option_map) - # 4 of SID, 8 of LSR and its padding, 8 of SEC and its padding, 4 of - # MTUP, with the NOP dropped. The SID half of this was 20 before #534, - # the extra four being the two octets it was over-wide by plus the two of - # padding they then needed. + # The parsed NOP is written in place, and the options are padded once, + # after the last: 1 of NOP, 4 of SID, 7 of LSR, 5 of SEC and 4 of MTUP + # make 21, so three octets of padding -- two NOPs and the EOOL, which + # keeps the multi-NOP arm of the alignment branch exercised on this, the + # data-model path. An EOOL after each unaligned option would end the + # list there on parse. C.f. #1316 and #1317. self.assertEqual(mapped_total, 24) - self.assertEqual(mapped_options[0].sid, 123) - # As above: the terminator is an EOOL option schema, so look for its type - # rather than for the wire code itself. See #506. - self.assertIn(OptionNumber.EOOL, [item.type for item in mapped_options]) - # Two NOPs from the SEC option's three octets of padding, which is the - # arm that ``for _ in range(pad_len - 1)`` only reaches when pad_len > 1. - self.assertEqual( - [item.type for item in mapped_options].count(OptionNumber.NOP), 2) - self.assertEqual(mapped_options[-1].mtu, 1500) + self.assertEqual([item.type for item in mapped_options], [ + OptionNumber.NOP, OptionNumber.SID, OptionNumber.LSR, OptionNumber.SEC, + OptionNumber.MTUP, OptionNumber.NOP, OptionNumber.NOP, OptionNumber.EOOL, + ]) + self.assertEqual(mapped_options[1].sid, 123) + self.assertEqual(mapped_options[4].mtu, 1500) def test_ipv4_option_readers_cover_common_and_error_branches(self) -> None: from pcapkit.const.ipv4.classification_level import ClassificationLevel @@ -1700,7 +1686,7 @@ def assert_bad(reader, schema) -> None: sec = proto._read_opt_sec( SECOption(type=OptionNumber.SEC, length=4, - level=ClassificationLevel.Unclassified, data=b'\x81'), + level=ClassificationLevel.Unclassified, data=b'\x80'), options=options, ) self.assertEqual(sec.level, ClassificationLevel.Unclassified) @@ -1714,7 +1700,7 @@ def assert_bad(reader, schema) -> None: proto._read_opt_sec( SECOption(type=OptionNumber.SEC, length=5, level=ClassificationLevel.Unclassified, - data=b'\x05\x00'), + data=b'\x04\x01'), options=options, ) self.assertEqual(warn.call_count, 3) @@ -1779,7 +1765,7 @@ def assert_bad(reader, schema) -> None: length=8, flags={'func': QSFunction.Quick_Start_Request, 'rate': 1}, ttl=7, - nonce={'nonce': 3}, + nonce={'nonce': 3, 'reserved': 0}, ) object.__setattr__(qs_request, 'func', QSFunction.Quick_Start_Request) qs_req = proto._read_opt_qs(qs_request, options=options) @@ -1790,7 +1776,8 @@ def assert_bad(reader, schema) -> None: type=OptionNumber.QS, length=8, flags={'func': QSFunction.Report_of_Approved_Rate, 'rate': 1}, - nonce={'nonce': 3}, + reserved=0, + nonce={'nonce': 3, 'reserved': 0}, ) object.__setattr__(qs_report, 'func', QSFunction.Report_of_Approved_Rate) qs_rep = proto._read_opt_qs(qs_report, options=options) @@ -1815,7 +1802,7 @@ def assert_bad(reader, schema) -> None: proto._read_ipv4_options(3) assert_bad(proto._read_opt_unassigned, UnassignedOption(type=OptionNumber.get(31), - length=2, data=b'')) + length=1, data=b'')) assert_bad(proto._read_opt_sec, SECOption(type=OptionNumber.SEC, length=2, level=ClassificationLevel.Unclassified, data=b'')) @@ -1891,8 +1878,9 @@ def test_ipv4_schema_helpers_and_post_process_branches(self) -> None: 'flags': {'func': QSFunction.Report_of_Approved_Rate.value}, }) self.assertIs(report_field.schema, ipv4_schema.QuickStartReportOption) - with self.assertRaises(FieldValueError): - ipv4_schema.quick_start_data_selector({'flags': {'func': 1}}) + unassigned_field = ipv4_schema.quick_start_data_selector({'flags': {'func': 1, 'length': 8}}) + self.assertIs(unassigned_field.schema, ipv4_schema.UnassignedOption) + self.assertEqual(unassigned_field.length, 8) wrapped_qs = ipv4_schema.QuickStartReportOption( type=OptionNumber.QS, @@ -1919,8 +1907,8 @@ def test_ipv4_schema_helpers_and_post_process_branches(self) -> None: self.assertEqual(ts_only.ts_flag, TSFlag.Timestamp_Only) self.assertEqual(ts_only.data, [1000, 0x80000005]) self.assertEqual(ts_only.timestamp[0], datetime.timedelta(seconds=1)) - self.assertEqual(ts_only.timestamp[1], 5) - warn.assert_called_once() + self.assertEqual(ts_only.timestamp[1], 0x80000005) + warn.assert_not_called() ip_ts = ipv4_schema.TSOption( type=OptionNumber.TS, @@ -1939,8 +1927,8 @@ def test_ipv4_schema_helpers_and_post_process_branches(self) -> None: self.assertIsInstance(ip_ts.data, OrderedMultiDict) self.assertEqual(ip_ts.timestamp[ip_address('192.0.2.1')], datetime.timedelta(seconds=2)) - self.assertEqual(ip_ts.timestamp[ip_address('192.0.2.4')], 7) - warn.assert_called_once() + self.assertEqual(ip_ts.timestamp[ip_address('192.0.2.4')], 0x80000007) + warn.assert_not_called() pre_ts = ipv4_schema.TSOption( type=OptionNumber.TS, @@ -1957,11 +1945,11 @@ def test_ipv4_schema_helpers_and_post_process_branches(self) -> None: pre_ts.remainder = ip_address('192.0.2.3').packed + b'\x00\x00\x00\x00' with mock.patch('pcapkit.protocols.schema.internet.ipv4.warn') as warn: pre_ts.post_process({}) - self.assertEqual(pre_ts.timestamp[ip_address('192.0.2.2')], 6) + self.assertEqual(pre_ts.timestamp[ip_address('192.0.2.2')], 0x80000006) self.assertEqual(pre_ts.timestamp[ip_address('192.0.2.5')], datetime.timedelta(seconds=3)) self.assertEqual(pre_ts.data[ip_address('192.0.2.3')], 0) - warn.assert_called_once() + warn.assert_not_called() unknown = ipv4_schema.TSOption( type=OptionNumber.TS,