diff --git a/pcapkit/protocols/data/internet/hip.py b/pcapkit/protocols/data/internet/hip.py index 081733912..5170f35d3 100644 --- a/pcapkit/protocols/data/internet/hip.py +++ b/pcapkit/protocols/data/internet/hip.py @@ -508,10 +508,22 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', suit @info_final class CertParameter(Parameter): - """Data model for HIP ``CERT`` parameter.""" + """Data model for HIP ``CERT`` parameter. - #: Certificate group. - cert_group: 'Group' + .. versionchanged:: 1.5.0 + + :attr:`cert_group` is an :obj:`int` rather than a + :class:`~pcapkit.const.hip.group.Group` (:issue:`2283`). + + """ + + #: Certificate group -- :rfc:`8002#section-2`'s "Group ID grouping + #: multiple related CERT parameters". An arbitrary number, since that + #: section leaves the CERT group namespace to each sending host, so no + #: registry stands behind it; it used to decode through the + #: Diffie-Hellman group registry and report a group of ``7`` as the curve + #: ``NIST_P_256``. See :issue:`2283`. + cert_group: 'int' #: Certificate count. cert_count: 'int' #: Certificate ID. @@ -522,7 +534,7 @@ class CertParameter(Parameter): cert: 'bytes' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', cert_group: 'Group', cert_count: 'int', cert_id: 'int', cert_type: 'Certificate', cert: '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', cert_group: 'int', cert_count: 'int', cert_id: 'int', cert_type: 'Certificate', cert: 'bytes') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -559,20 +571,45 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', opaq @info_final class Lifetime(Data): - """Data model for registration lifetime.""" + """Data model for registration lifetime. - #: Minimum lifetime. + .. versionchanged:: 1.5.0 + + New required fields ``min_exponent`` and ``max_exponent``, and + :attr:`min` and :attr:`max` follow :rfc:`8003#section-4.1`'s exponential + encoding rather than reading the octet as a count of seconds + (:issue:`2281`). + + """ + + #: The ``Min Lifetime`` octet as it appears on the wire. It is what + #: re-serialising writes back, since most of the 256 octets have no exact + #: :class:`~datetime.timedelta`. + min_exponent: 'int' + #: The ``Max Lifetime`` octet as it appears on the wire. + max_exponent: 'int' + #: Minimum lifetime, decoded from :attr:`min_exponent` per + #: :rfc:`8003#section-4.1`. min: 'timedelta' - #: Maximum lifetime. + #: Maximum lifetime, decoded from :attr:`max_exponent` per + #: :rfc:`8003#section-4.1`. max: 'timedelta' if TYPE_CHECKING: - def __init__(self, min: 'timedelta', max: 'timedelta') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long + def __init__(self, min_exponent: 'int', max_exponent: 'int', min: 'timedelta', max: 'timedelta') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final class RegInfoParameter(Parameter): - """Data model for HIP ``REG_INFO`` parameter.""" + """Data model for HIP ``REG_INFO`` parameter. + + .. versionchanged:: 1.5.0 + + :attr:`lifetime` carries the raw octets alongside the durations, and the + durations follow :rfc:`8003#section-4.1` -- see + :class:`Lifetime` (:issue:`2281`). + + """ #: Registration lifetime. lifetime: 'Lifetime' @@ -585,28 +622,55 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', life @info_final class RegRequestParameter(Parameter): - """Data model for HIP ``REG_REQUEST`` parameter.""" + """Data model for HIP ``REG_REQUEST`` parameter. - #: Registration lifetime. + .. versionchanged:: 1.5.0 + + New required field ``lifetime_exponent``, and :attr:`lifetime` follows + :rfc:`8003#section-4.1`'s exponential encoding rather than reading the + octet as a count of seconds (:issue:`2281`). + + """ + + #: The ``Lifetime`` octet as it appears on the wire. :rfc:`8003#section-4.1` + #: makes a non-zero octet an exponent standing for ``2^((lifetime - 64)/8)`` + #: seconds, and this is that octet. It is what re-serialising writes back, + #: since most of the 256 octets have no exact + #: :class:`~datetime.timedelta`. + lifetime_exponent: 'int' + #: Registration lifetime, decoded from :attr:`lifetime_exponent` per + #: :rfc:`8003#section-4.1`. lifetime: 'timedelta' #: Registration type. reg_type: 'tuple[Registration, ...]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', lifetime: 'timedelta', reg_type: 'tuple[Registration, ...]') -> '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', lifetime_exponent: 'int', lifetime: 'timedelta', reg_type: 'tuple[Registration, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final class RegResponseParameter(Parameter): - """Data model for HIP ``REG_RESPONSE`` parameter.""" + """Data model for HIP ``REG_RESPONSE`` parameter. - #: Registration lifetime. + .. versionchanged:: 1.5.0 + + New required field ``lifetime_exponent``, and :attr:`lifetime` follows + :rfc:`8003#section-4.1`'s exponential encoding rather than reading the + octet as a count of seconds (:issue:`2281`). + + """ + + #: The ``Lifetime`` octet as it appears on the wire, for the reason + #: :attr:`RegRequestParameter.lifetime_exponent` gives. + lifetime_exponent: 'int' + #: Registration lifetime, decoded from :attr:`lifetime_exponent` per + #: :rfc:`8003#section-4.1`. lifetime: 'timedelta' #: Registration type. reg_type: 'tuple[Registration, ...]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', lifetime: 'timedelta', reg_type: 'tuple[Registration, ...]') -> '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', lifetime_exponent: 'int', lifetime: 'timedelta', reg_type: 'tuple[Registration, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -743,24 +807,48 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', rese @info_final class TransactionIDParameter(Parameter): - """Data model for HIP ``TRANSACTION_ID`` parameter.""" + """Data model for HIP ``TRANSACTION_ID`` parameter. + + .. versionchanged:: 1.5.0 + + New required field ``id_len`` (:issue:`2271`). + + """ #: Identifier. id: 'int' + #: On-wire width of :attr:`id`, in octets -- the parameter's own ``Length``, + #: which :rfc:`6078#section-4.4` gives as the "Length of the Identifier, in + #: octets" for what is otherwise an opaque value. Declared rather than + #: recomputed from :meth:`int.bit_length`, which cannot see a leading zero + #: octet: without it a parameter read with ``Length = 4`` re-serialised as + #: ``Length = 1``. Same class of bug as :issue:`653`; see :issue:`2271`. + id_len: 'int' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', id: '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', id: 'int', id_len: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final class OverlayIDParameter(Parameter): - """Data model for HIP ``OVERLAY_ID`` parameter.""" + """Data model for HIP ``OVERLAY_ID`` parameter. + + .. versionchanged:: 1.5.0 + + New required field ``id_len`` (:issue:`2271`). + + """ #: Identifier. id: 'int' + #: On-wire width of :attr:`id`, in octets -- the parameter's own ``Length``, + #: which :rfc:`6079#section-6.1` gives as the "Length of the Identifier, in + #: octets" for what is otherwise an opaque value. Declared for the reason + #: :attr:`TransactionIDParameter.id_len` gives. See :issue:`2271`. + id_len: 'int' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', id: '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', id: 'int', id_len: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final @@ -941,17 +1029,22 @@ class OverlayTTLParameter(Parameter): .. versionchanged:: 1.5.0 - New required field ``reserved`` (:issue:`1223`). + New required field ``reserved`` (:issue:`1223`), and :attr:`ttl` is an + :obj:`int` rather than a :class:`~datetime.timedelta` (:issue:`2282`). """ #: The two reserved octets, as captured. reserved: 'bytes' - #: TTL value. - ttl: 'timedelta' + #: TTL value -- "the number of overlay hops this packet has left" + #: (:rfc:`6079#section-6.2`), decremented by one per overlay hop. A count, + #: not a duration, so it is a plain :obj:`int`: reading it as + #: ``timedelta(seconds=ttl)`` reported a 5-hop budget as five seconds. + #: See :issue:`2282`. + ttl: 'int' if TYPE_CHECKING: - 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 + def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'bytes', ttl: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long @info_final diff --git a/pcapkit/protocols/internet/hip.py b/pcapkit/protocols/internet/hip.py index 3c462e994..0c628dcb1 100644 --- a/pcapkit/protocols/internet/hip.py +++ b/pcapkit/protocols/internet/hip.py @@ -1869,6 +1869,117 @@ def _read_param_echo_request_signed(self, schema: 'Schema_EchoRequestSignedParam ) return echo_request_signed + @staticmethod + def _read_registration_lifetime(octet: 'int') -> 'timedelta': + """Decode a registration ``Lifetime`` octet into a duration. + + :rfc:`8003#section-4.1` encodes the registration lifetimes of + ``REG_INFO``, ``REG_REQUEST`` and ``REG_RESPONSE`` as an exponent: the + octet ``0`` is "a special lifetime duration of 0 (zero) seconds", used + to request and grant cancellation, and any other value stands for + ``2^((lifetime - 64)/8)`` seconds. That spans ``2^(-63/8)`` seconds + (about 4 ms) to ``2^(191/8)`` seconds (about 178 days), all of which + :class:`~datetime.timedelta` holds, so -- unlike + :meth:`_read_puzzle_lifetime` -- no octet overflows and the return is + never :obj:`None`. + + Reading the octet as a count of seconds instead reported every duration + wrongly: an octet of ``64``, which is one second, read as 64 seconds. + See :issue:`2281`. + + Args: + octet: the ``Lifetime`` octet + + Returns: + The lifetime. + + """ + if not octet: + return datetime.timedelta(0) + return datetime.timedelta(seconds=2 ** ((octet - 64) / 8)) + + @staticmethod + def _make_registration_lifetime(code: 'Enum_Parameter', version: 'int', + lifetime: 'timedelta | int | float') -> 'int': + """Encode a registration ``Lifetime`` octet. + + The inverse of :meth:`_read_registration_lifetime`, rounding down to the + nearest representable octet as :meth:`_make_puzzle_lifetime` does for + ``PUZZLE``. A duration of zero is the one exact case: + :rfc:`8003#section-4.1` reserves the octet ``0`` for it. + + Args: + code: parameter code + version: HIP protocol version + lifetime: lifetime, as a :class:`~datetime.timedelta` or a number of + seconds + + Returns: + The ``Lifetime`` octet. + + Raises: + ProtocolError: If ``lifetime`` is a negative duration, or encodes to + a value the one-octet field cannot hold. + + """ + # Keyed on :class:`~datetime.timedelta`, not on :class:`int`: testing + # for ``int`` would send a plain ``float`` down the timedelta branch and + # escape an :exc:`AttributeError`. + seconds = lifetime.total_seconds() if isinstance(lifetime, timedelta) else lifetime + + if seconds < 0: + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid lifetime: ' + f'{seconds} is not a non-negative number of seconds') + # Zero is the RFC's own special case, so it is encoded rather than + # rejected -- which is where this parts company with + # ``_make_puzzle_lifetime``, whose field has no such value. Written as a + # truth test rather than ``== 0`` because ``seconds`` may be a float. + if not seconds: + return 0 + + octet = math.floor(math.log2(seconds) * 8 + 64) + + # ``UInt8Field`` wraps rather than raising (``300`` packs as ``0x2c``), so + # an out-of-range lifetime would otherwise be written as some other + # valid-looking duration. + if not 1 <= octet <= 0xFF: + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid lifetime: ' + f'{seconds} seconds encodes to {octet}, outside the ' + 'one-octet Lifetime field') + return octet + + def _make_lifetime_octet(self, code: 'Enum_Parameter', version: 'int', + exponent: 'Optional[int]', + lifetime: 'timedelta | int | float') -> 'int': + """Resolve a registration ``Lifetime`` octet from either input form. + + An explicit ``exponent`` is the octet and wins, as a parsed parameter's + is; otherwise ``lifetime`` is a duration and + :meth:`_make_registration_lifetime` encodes it. Shared by the three + :rfc:`8003` makers, which between them carry four such octets. + + Args: + code: parameter code + version: HIP protocol version + exponent: the ``Lifetime`` octet itself, if any + lifetime: lifetime, as a :class:`~datetime.timedelta` or a number of + seconds + + Returns: + The ``Lifetime`` octet. + + Raises: + ProtocolError: If ``exponent`` does not fit one octet, or + ``lifetime`` cannot be encoded. + + """ + if exponent is None: + return self._make_registration_lifetime(code, version, lifetime) + if not 0 <= exponent <= 0xFF: + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid lifetime: ' + f'{exponent} is outside the one-octet Lifetime field') + return exponent + def _read_param_reg_info(self, schema: 'Schema_RegInfoParameter', *, version: 'int', # pylint: disable=unused-argument options: 'Parameter') -> 'Data_RegInfoParameter': # pylint: disable=unused-argument """Read HIP ``REG_INFO`` parameter. @@ -1903,8 +2014,12 @@ def _read_param_reg_info(self, schema: 'Schema_RegInfoParameter', *, version: 'i critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), lifetime=Data_Lifetime( - min=datetime.timedelta(seconds=schema.min_lifetime), - max=datetime.timedelta(seconds=schema.max_lifetime), + # The octets are the round-trip source; the durations are + # decoded from them per :rfc:`8003#section-4.1` (#2281). + min_exponent=schema.min_lifetime, + max_exponent=schema.max_lifetime, + min=self._read_registration_lifetime(schema.min_lifetime), + max=self._read_registration_lifetime(schema.max_lifetime), ), reg_type=tuple(schema.reg_info), ) @@ -1943,7 +2058,8 @@ def _read_param_reg_request(self, schema: 'Schema_RegRequestParameter', *, versi type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), - lifetime=datetime.timedelta(seconds=schema.lifetime), + lifetime_exponent=schema.lifetime, + lifetime=self._read_registration_lifetime(schema.lifetime), reg_type=tuple(schema.reg_request), ) return reg_request @@ -1981,7 +2097,8 @@ def _read_param_reg_response(self, schema: 'Schema_RegResponseParameter', *, ver type=schema.type, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), - lifetime=datetime.timedelta(seconds=schema.lifetime), + lifetime_exponent=schema.lifetime, + lifetime=self._read_registration_lifetime(schema.lifetime), reg_type=tuple(schema.reg_response), ) return reg_response @@ -2346,6 +2463,10 @@ def _read_param_transaction_id(self, schema: 'Schema_TransactionIDParameter', *, critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), id=schema.id, + # Keep the declared width, which ``schema.id`` alone cannot carry + # and which ``length`` cannot recover: ``parameter_total_len`` + # pads, so ``Length`` 1 through 4 all give 8 (#2271). + id_len=schema.len, ) return transaction_id @@ -2381,6 +2502,9 @@ def _read_param_overlay_id(self, schema: 'Schema_OverlayIDParameter', *, version critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), id=schema.id, + # The declared width, for the reason ``_read_param_transaction_id`` + # gives (#2271). + id_len=schema.len, ) return overlay_id @@ -2834,7 +2958,9 @@ def _read_param_overlay_ttl(self, schema: 'Schema_OverlayTTLParameter', *, versi critical=bool(schema.type & 0b1), length=parameter_total_len(schema.len), reserved=schema.reserved, - ttl=datetime.timedelta(seconds=schema.ttl), + # A hop count, not a duration -- see ``OverlayTTLParameter.ttl`` + # in :mod:`pcapkit.protocols.data.internet.hip` (#2282). + ttl=schema.ttl, ) return overlay_ttl @@ -3283,7 +3409,7 @@ def _make_param_locator_set(self, code: 'Enum_Parameter', param: 'Optional[Data_ """ def _make_locator(locator: 'Optional[Data_Locator]' = None, *, traffic: 'int' = 0, - type: 'int' = 0, + type: 'Optional[int]' = None, preferred: 'bool' = False, reserved: 'int' = 0, lifetime: 'timedelta | int' = 0, @@ -3295,7 +3421,8 @@ def _make_locator(locator: 'Optional[Data_Locator]' = None, *, Args: locator: locator data traffic: traffic type - type: locator type + type: locator type; derived from ``spi`` when not given -- + ``1`` with an ``spi``, ``0`` without preferred: preferred flag reserved: the 7 reserved high-order bits of the flags octet lifetime: lifetime @@ -3306,6 +3433,12 @@ def _make_locator(locator: 'Optional[Data_Locator]' = None, *, Returns: HIP locator schema. + Raises: + ProtocolError: If ``locator`` holds neither an + :class:`~ipaddress.IPv6Address` nor a + :class:`~pcapkit.protocols.data.internet.hip.LocatorData`, + or if an explicit ``type`` contradicts ``spi``. + """ if locator is not None: value = locator.locator @@ -3338,6 +3471,21 @@ def _make_locator(locator: 'Optional[Data_Locator]' = None, *, ip_val = parse_ip_address( ip, f'HIPv{version}: [ParamNo {code}] invalid locator', version=6) + # ``Locator Type`` and the presence of an SPI are one fact + # written twice: :rfc:`8046#section-4.2` makes Type 0 a bare + # IPv6 address and Type 1 "the concatenation of an ESP SPI + # (first 32 bits) followed by an IPv6 address". ``Locator + # Length`` follows from the choice, and ``locator_value_selector`` + # accepts only the two consistent pairs, so leaving ``type`` at + # a default of 0 while an ``spi`` set the length to 5 made the + # public ``spi=`` path raise (#2276). Deriving it keeps + # the caller from restating what ``spi`` already says. + if type is None: + type = 0 if spi is None else 1 + elif (type == 1) != (spi is not None): + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid locator type: ' + f'{type!r} does not match spi {spi!r}') + if spi is None: length = 4 data = ip_val.packed @@ -3820,7 +3968,18 @@ def _make_param_hip_transform(self, code: 'Enum_Parameter', param: 'Optional[Dat Returns: HIP parameter schema. + Raises: + ProtocolError: The parameter is **NOT** supported outside HIPv1. + """ + # ``HIP_TRANSFORM`` is :rfc:`5201#section-5.2.7` only -- :rfc:`7401` + # replaced it with ``HIP_CIPHER`` -- and the gate mirrors + # :meth:`_read_param_hip_transform`'s. Without it the maker built + # octets its own reader rejects (#2275), where + # :meth:`_make_param_hip_cipher` refuses the opposite mismatch. + if version != 1: + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid parameter') + if param is not None: suite_id = cast('list[Enum_Suite]', param.suite_id) else: @@ -4105,7 +4264,7 @@ def _make_param_host_id(self, code: 'Enum_Parameter', param: 'Optional[Data_Host ) algorithm = Enum_HIAlgorithm.EdDSA else: - raise ProtocolError(f'[HIPv{version}] invalid curve: {hi_curve!r}') + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid curve: {hi_curve!r}') else: if isinstance(hi, Data_HostIdentity): hi_len = 2 + len(hi.pubkey) @@ -4128,7 +4287,7 @@ def _make_param_host_id(self, code: 'Enum_Parameter', param: 'Optional[Data_Host ) algorithm = Enum_HIAlgorithm.EdDSA else: - raise ProtocolError(f'[HIPv{version}] invalid curve: {hi.curve!r}') + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid curve: {hi.curve!r}') elif isinstance(hi, Schema_HostIdentity): hi_len = 2 + len(hi.pub_key) hi_data = hi @@ -4197,10 +4356,7 @@ def _make_param_hit_suite_list(self, code: 'Enum_Parameter', param: 'Optional[Da def _make_param_cert(self, code: 'Enum_Parameter', param: 'Optional[Data_CertParameter]' = None, *, # pylint: disable=unused-argument version: 'int', - cert_group: 'Enum_Group | StdlibEnum | AenumEnum | str | int' = Enum_Group.NIST_P_256, - cert_group_default: 'Optional[int]' = None, - cert_group_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long - cert_group_reversed: 'bool' = False, + cert_group: 'int' = 0, cert_count: 'int' = 0, cert_id: 'int' = 0, cert_type: 'Enum_Certificate | StdlibEnum | AenumEnum | str | int' = Enum_Certificate.X_509_v3, @@ -4215,10 +4371,8 @@ def _make_param_cert(self, code: 'Enum_Parameter', param: 'Optional[Data_CertPar code: parameter code param: parameter data version: HIP protocol version - cert_group: certificate group - cert_group_default: default certificate group - cert_group_namespace: certificate group namespace - cert_group_reversed: reverse certificate group namespace + cert_group: certificate group ID -- an arbitrary number, locally + managed by the sender per :rfc:`8002#section-2` cert_count: certificate count cert_id: certificate ID cert_type: certificate type @@ -4231,6 +4385,14 @@ def _make_param_cert(self, code: 'Enum_Parameter', param: 'Optional[Data_CertPar Returns: HIP parameter schema. + .. versionchanged:: 1.5.0 + + ``cert_group`` is a plain :obj:`int` and defaults to ``0``; the + ``cert_group_default``, ``cert_group_namespace`` and + ``cert_group_reversed`` arguments are gone, since they existed only + to look the value up in the Diffie-Hellman group registry that + :rfc:`8002` does not define for it (:issue:`2283`). + """ if param is not None: group = param.cert_group @@ -4239,8 +4401,7 @@ def _make_param_cert(self, code: 'Enum_Parameter', param: 'Optional[Data_CertPar type = param.cert_type cert = param.cert else: - group = self._make_index(cert_group, cert_group_default, namespace=cert_group_namespace, # type: ignore[assignment] - reversed=cert_group_reversed, pack=False) + group = cert_group count = cert_count id = cert_id type = self._make_index(cert_type, cert_type_default, namespace=cert_type_namespace, # type: ignore[assignment] @@ -4328,8 +4489,10 @@ def _make_param_echo_request_signed(self, code: 'Enum_Parameter', param: 'Option def _make_param_reg_info(self, code: 'Enum_Parameter', param: 'Optional[Data_RegInfoParameter]' = None, *, # pylint: disable=unused-argument version: 'int', - min_lifetime: 'int | timedelta' = 0, - max_lifetime: 'int | timedelta' = 0xf, + min_lifetime: 'timedelta | int | float' = 0, + max_lifetime: 'timedelta | int | float' = 0xf, + min_lifetime_exponent: 'Optional[int]' = None, + max_lifetime_exponent: 'Optional[int]' = None, reg_info: 'Optional[list[Enum_Registration | StdlibEnum | AenumEnum | str | int]]' = None, reg_info_default: 'Optional[int]' = None, reg_info_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long @@ -4341,8 +4504,16 @@ def _make_param_reg_info(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg code: parameter code param: parameter data version: HIP protocol version - min_lifetime: minimum lifetime - max_lifetime: maximum lifetime + min_lifetime: minimum lifetime, encoded to the nearest lower + ``Min Lifetime`` octet; ignored when + ``min_lifetime_exponent`` is given + max_lifetime: maximum lifetime, encoded to the nearest lower + ``Max Lifetime`` octet; ignored when + ``max_lifetime_exponent`` is given + min_lifetime_exponent: the ``Min Lifetime`` octet itself, + ``lifetime`` in :rfc:`8003#section-4.1`'s + ``2^((lifetime - 64)/8)`` seconds + max_lifetime_exponent: the ``Max Lifetime`` octet itself reg_info: registration information list reg_info_default: default registration information reg_info_namespace: registration information namespace @@ -4352,11 +4523,25 @@ def _make_param_reg_info(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg Returns: HIP parameter schema. + Raises: + ProtocolError: If either exponent does not fit one octet, or either + lifetime cannot be encoded; see + ``_make_registration_lifetime``. + + .. versionchanged:: 1.5.0 + + ``min_lifetime`` and ``max_lifetime`` are durations encoded per + :rfc:`8003#section-4.1` rather than octets written verbatim, and the + new ``min_lifetime_exponent`` and ``max_lifetime_exponent`` take the + octets directly (:issue:`2281`). + """ if param is not None: reg_type = cast('list[Enum_Registration]', param.reg_type) - min_lt = math.floor(param.lifetime.min.total_seconds()) - max_lt = math.floor(param.lifetime.max.total_seconds()) + if min_lifetime_exponent is None: + min_lifetime_exponent = param.lifetime.min_exponent + if max_lifetime_exponent is None: + max_lifetime_exponent = param.lifetime.max_exponent else: if reg_info is None: reg_info = [] @@ -4366,8 +4551,8 @@ def _make_param_reg_info(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg reg_type.append(self._make_index(reg, reg_info_default, namespace=reg_info_namespace, # type: ignore[arg-type] reversed=reg_info_reversed, pack=False)) - min_lt = min_lifetime if isinstance(min_lifetime, int) else math.floor(min_lifetime.total_seconds()) - max_lt = max_lifetime if isinstance(max_lifetime, int) else math.floor(max_lifetime.total_seconds()) + min_lt = self._make_lifetime_octet(code, version, min_lifetime_exponent, min_lifetime) + max_lt = self._make_lifetime_octet(code, version, max_lifetime_exponent, max_lifetime) return Schema_RegInfoParameter( type=code, @@ -4379,7 +4564,8 @@ def _make_param_reg_info(self, code: 'Enum_Parameter', param: 'Optional[Data_Reg def _make_param_reg_request(self, code: 'Enum_Parameter', param: 'Optional[Data_RegRequestParameter]' = None, *, # pylint: disable=unused-argument version: 'int', - lifetime: 'int | timedelta' = 0, + lifetime: 'timedelta | int | float' = 0, + lifetime_exponent: 'Optional[int]' = None, reg_request: 'Optional[list[Enum_Registration | StdlibEnum | AenumEnum | str | int]]' = None, reg_request_default: 'Optional[int]' = None, reg_request_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long @@ -4391,7 +4577,10 @@ def _make_param_reg_request(self, code: 'Enum_Parameter', param: 'Optional[Data_ code: parameter code param: parameter data version: HIP protocol version - lifetime: lifetime + lifetime: lifetime, encoded to the nearest lower ``Lifetime`` + octet; ignored when ``lifetime_exponent`` is given + lifetime_exponent: the ``Lifetime`` octet itself, ``lifetime`` in + :rfc:`8003#section-4.1`'s ``2^((lifetime - 64)/8)`` seconds reg_request: registration request list reg_request_default: default registration request reg_request_namespace: registration request namespace @@ -4401,10 +4590,22 @@ def _make_param_reg_request(self, code: 'Enum_Parameter', param: 'Optional[Data_ Returns: HIP parameter schema. + Raises: + ProtocolError: If ``lifetime_exponent`` does not fit one octet, or + ``lifetime`` cannot be encoded; see + ``_make_registration_lifetime``. + + .. versionchanged:: 1.5.0 + + ``lifetime`` is a duration encoded per :rfc:`8003#section-4.1` + rather than an octet written verbatim, and the new + ``lifetime_exponent`` takes the octet directly (:issue:`2281`). + """ if param is not None: reg_type = cast('list[Enum_Registration]', param.reg_type) - lt = math.floor(param.lifetime.total_seconds()) + if lifetime_exponent is None: + lifetime_exponent = param.lifetime_exponent else: if reg_request is None: reg_request = [] @@ -4414,7 +4615,7 @@ def _make_param_reg_request(self, code: 'Enum_Parameter', param: 'Optional[Data_ reg_type.append(self._make_index(reg, reg_request_default, namespace=reg_request_namespace, # type: ignore[arg-type] reversed=reg_request_reversed, pack=False)) - lt = lifetime if isinstance(lifetime, int) else math.floor(lifetime.total_seconds()) + lt = self._make_lifetime_octet(code, version, lifetime_exponent, lifetime) return Schema_RegRequestParameter( type=code, @@ -4425,7 +4626,8 @@ def _make_param_reg_request(self, code: 'Enum_Parameter', param: 'Optional[Data_ def _make_param_reg_response(self, code: 'Enum_Parameter', param: 'Optional[Data_RegResponseParameter]' = None, *, # pylint: disable=unused-argument version: 'int', - lifetime: 'int | timedelta' = 0, + lifetime: 'timedelta | int | float' = 0, + lifetime_exponent: 'Optional[int]' = None, reg_response: 'Optional[list[Enum_Registration | StdlibEnum | AenumEnum | str | int]]' = None, reg_response_default: 'Optional[int]' = None, reg_response_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long @@ -4437,7 +4639,10 @@ def _make_param_reg_response(self, code: 'Enum_Parameter', param: 'Optional[Data code: parameter code param: parameter data version: HIP protocol version - lifetime: lifetime + lifetime: lifetime, encoded to the nearest lower ``Lifetime`` + octet; ignored when ``lifetime_exponent`` is given + lifetime_exponent: the ``Lifetime`` octet itself, ``lifetime`` in + :rfc:`8003#section-4.1`'s ``2^((lifetime - 64)/8)`` seconds reg_response: registration response list reg_response_default: default registration response reg_response_namespace: registration response namespace @@ -4447,10 +4652,22 @@ def _make_param_reg_response(self, code: 'Enum_Parameter', param: 'Optional[Data Returns: HIP parameter schema. + Raises: + ProtocolError: If ``lifetime_exponent`` does not fit one octet, or + ``lifetime`` cannot be encoded; see + ``_make_registration_lifetime``. + + .. versionchanged:: 1.5.0 + + ``lifetime`` is a duration encoded per :rfc:`8003#section-4.1` + rather than an octet written verbatim, and the new + ``lifetime_exponent`` takes the octet directly (:issue:`2281`). + """ if param is not None: reg_type = cast('list[Enum_Registration]', param.reg_type) - lt = math.floor(param.lifetime.total_seconds()) + if lifetime_exponent is None: + lifetime_exponent = param.lifetime_exponent else: if reg_response is None: reg_response = [] @@ -4460,7 +4677,7 @@ def _make_param_reg_response(self, code: 'Enum_Parameter', param: 'Optional[Data reg_type.append(self._make_index(reg, reg_response_default, namespace=reg_response_namespace, # type: ignore[arg-type] reversed=reg_response_reversed, pack=False)) - lt = lifetime if isinstance(lifetime, int) else math.floor(lifetime.total_seconds()) + lt = self._make_lifetime_octet(code, version, lifetime_exponent, lifetime) return Schema_RegResponseParameter( type=code, @@ -4795,9 +5012,67 @@ def _make_param_payload_mic(self, code: 'Enum_Parameter', param: 'Optional[Data_ mic=mic, ) + @staticmethod + def _make_identifier_width(code: 'Enum_Parameter', version: 'int', + id_len: 'Optional[int]', identifier: 'int') -> 'int': + """Resolve the on-wire width of an opaque ``Identifier`` field. + + ``TRANSACTION_ID`` (:rfc:`6078#section-4.4`) and ``OVERLAY_ID`` + (:rfc:`6079#section-6.1`) both spell their ``Length`` as the "Length of + the Identifier, in octets" over an otherwise opaque value, so the width + is a property of the identifier as issued and not of the number that + happens to be in it. It is resolved here in that order of authority: + + 1. an explicit ``id_len`` argument, which is what a parsed parameter + carries on :attr:`TransactionIDParameter.id_len + `; + 2. failing that, the identifier's own :meth:`int.bit_length`, rounded up + to whole octets. + + Only case 2 can lose a leading zero octet, and it is the only case + where the width is genuinely unknowable -- a from-scratch build with + nothing declaring it. Using it unconditionally re-serialised a + ``Length = 4`` identifier of ``1`` as ``Length = 1`` (:issue:`2271`), + the same way :meth:`_make_puzzle_field_width` guards ``Random #I`` + against :issue:`653`. + + A width of ``0`` is accepted rather than rejected: it is what case 2 + yields for the default identifier of ``0``, and what a ``Length = 0`` + parameter parses back to, so refusing it would fail callers that pass no + value at all. + + Args: + code: parameter code + version: HIP protocol version + id_len: declared field width in octets, if any + identifier: the identifier that must fit + + Returns: + The width of the ``Identifier`` field, in octets. + + Raises: + ProtocolError: If the declared width is negative, too narrow for + ``identifier``, or beyond what the 2-octet ``Length`` field can + hold. + + """ + if id_len is None: + return math.ceil(identifier.bit_length() / 8) + + # ``UInt16Field`` wraps rather than raising, so an out-of-range width + # would otherwise be written as some other valid-looking ``Length``. + if not 0 <= id_len <= 0xFFFF: + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid identifier width: ' + f'{id_len} is outside the two-octet Length field') + if identifier.bit_length() > id_len * 8: + raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid identifier width: ' + f'{id_len} octets cannot hold {identifier!r}') + return id_len + def _make_param_transaction_id(self, code: 'Enum_Parameter', param: 'Optional[Data_TransactionIDParameter]' = None, *, # pylint: disable=unused-argument version: 'int', id: 'int' = 0, + id_len: 'Optional[int]' = None, **kwargs: 'Any') -> 'Schema_TransactionIDParameter': """Make HIP ``TRANSACTION_ID`` parameter. @@ -4806,24 +5081,38 @@ def _make_param_transaction_id(self, code: 'Enum_Parameter', param: 'Optional[Da param: parameter data version: HIP protocol version id: transaction ID + id_len: on-wire width of ``Identifier``, in octets; defaults to the + width ``param`` was parsed with, or to ``id``'s own bit length **kwargs: arbitrary keyword arguments Returns: HIP parameter schema. + Raises: + ProtocolError: If the width of ``Identifier`` is invalid; see + ``_make_identifier_width``. + + .. versionchanged:: 1.5.0 + + New ``id_len`` argument, so that an identifier with leading zero + octets rebuilds at the width it was read with (:issue:`2271`). + """ if param is not None: id = param.id + if id_len is None: + id_len = param.id_len return Schema_TransactionIDParameter( type=code, - len=math.ceil(id.bit_length() / 8), + len=self._make_identifier_width(code, version, id_len, id), id=id, ) def _make_param_overlay_id(self, code: 'Enum_Parameter', param: 'Optional[Data_OverlayIDParameter]' = None, *, # pylint: disable=unused-argument version: 'int', id: 'int' = 0, + id_len: 'Optional[int]' = None, **kwargs: 'Any') -> 'Schema_OverlayIDParameter': """Make HIP ``OVERLAY_ID`` parameter. @@ -4832,18 +5121,31 @@ def _make_param_overlay_id(self, code: 'Enum_Parameter', param: 'Optional[Data_O param: parameter data version: HIP protocol version id: overlay ID + id_len: on-wire width of ``Identifier``, in octets; defaults to the + width ``param`` was parsed with, or to ``id``'s own bit length **kwargs: arbitrary keyword arguments Returns: HIP parameter schema. + Raises: + ProtocolError: If the width of ``Identifier`` is invalid; see + ``_make_identifier_width``. + + .. versionchanged:: 1.5.0 + + New ``id_len`` argument, so that an identifier with leading zero + octets rebuilds at the width it was read with (:issue:`2271`). + """ if param is not None: id = param.id + if id_len is None: + id_len = param.id_len return Schema_OverlayIDParameter( type=code, - len=math.ceil(id.bit_length() / 8), + len=self._make_identifier_width(code, version, id_len, id), id=id, ) @@ -5220,7 +5522,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, + ttl: 'int' = 0, reserved: 'bytes' = b'\x00\x00', **kwargs: 'Any') -> 'Schema_OverlayTTLParameter': """Make HIP ``OVERLAY_TTL`` parameter. @@ -5229,7 +5531,10 @@ def _make_param_overlay_ttl(self, code: 'Enum_Parameter', param: 'Optional[Data_ code: parameter code param: parameter data version: HIP protocol version - ttl: overlay time-to-live (TTL) value + ttl: overlay time-to-live (TTL), as a count of remaining overlay + hops rather than a duration -- see + :attr:`OverlayTTLParameter.ttl + ` reserved: the two reserved octets, truncated or zero-filled to 2 octets **kwargs: arbitrary keyword arguments @@ -5237,12 +5542,17 @@ def _make_param_overlay_ttl(self, code: 'Enum_Parameter', param: 'Optional[Data_ Returns: HIP parameter schema. + .. versionchanged:: 1.5.0 + + ``ttl`` is a hop count rather than a duration, and no longer accepts + a :class:`~datetime.timedelta` (:issue:`2282`). + """ if param is not None: reserved = getattr(param, 'reserved', b'\x00\x00') - ttl_val = math.floor(param.ttl.total_seconds()) + ttl_val = param.ttl else: - ttl_val = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds()) + ttl_val = ttl return Schema_OverlayTTLParameter( reserved=reserved, diff --git a/pcapkit/protocols/schema/internet/hip.py b/pcapkit/protocols/schema/internet/hip.py index 29c624b2a..d90abc54d 100644 --- a/pcapkit/protocols/schema/internet/hip.py +++ b/pcapkit/protocols/schema/internet/hip.py @@ -1060,8 +1060,14 @@ def __init__(self, type: 'Enum_Parameter', len: 'int', suites: 'list[Enum_HITSui class CertParameter(Parameter, code=Enum_Parameter.CERT): """Header schema for HIP ``CERT`` parameters.""" - #: Certificate group. - cert_group: 'Enum_Group' = EnumField(length=1, namespace=Enum_Group) + #: Certificate group -- the "Group ID grouping multiple related CERT + #: parameters" of :rfc:`8002#section-2`. A plain octet, not a registry + #: value: that section puts the CERT group namespace under local + #: management by the sending host, so there is nothing to look it up in. + #: It was an :class:`~pcapkit.corekit.fields.numbers.EnumField` over the + #: Diffie-Hellman group registry, which read a group of ``7`` back as the + #: curve name ``NIST_P_256``. See :issue:`2283`. + cert_group: 'int' = UInt8Field() #: Certificate count. cert_count: 'int' = UInt8Field() #: Certificate ID. @@ -1074,7 +1080,7 @@ class CertParameter(Parameter, code=Enum_Parameter.CERT): padding: 'bytes' = PaddingField(length=parameter_padding_len) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Parameter', len: 'int', cert_group: 'Enum_Group', cert_count: 'int', + def __init__(self, type: 'Enum_Parameter', len: 'int', cert_group: 'int', cert_count: 'int', cert_id: 'int', cert_type: 'Enum_Certificate', cert: 'bytes') -> 'None': ... diff --git a/tests/project/test_rfc_constants_one_home_1739_unit.py b/tests/project/test_rfc_constants_one_home_1739_unit.py index 5dfa4bd9d..e5f139f47 100644 --- a/tests/project/test_rfc_constants_one_home_1739_unit.py +++ b/tests/project/test_rfc_constants_one_home_1739_unit.py @@ -353,7 +353,11 @@ 'pcapkit/protocols/application/ospf.py': ((24, 65535), 'OSPF header length and field bounds'), 'pcapkit/protocols/schema/application/ospf.py': ((24,), 'the OSPF header length'), 'pcapkit/protocols/internet/esp.py': ((12, 16, 20, 24, 32), 'ESP ICV and key sizes'), - 'pcapkit/protocols/internet/hip.py': ((6, 12, 16, 20, 32, 40), 'HIP parameter fields and sizes'), + 'pcapkit/protocols/internet/hip.py': ((6, 12, 16, 20, 32, 40, 65535), + 'HIP parameter fields and sizes; 65535 bounds the ' + 'two-octet TLV Length a TRANSACTION_ID or OVERLAY_ID ' + 'may declare (#2271), as it does in the schema module, ' + 'and is not an IP total length'), 'pcapkit/protocols/schema/internet/hip.py': ((12, 16, 20, 65535), 'HIP parameter fields, sizes and bit-fields'), 'pcapkit/protocols/internet/hopopt.py': ((6, 16, 32), 'HOPOPT option fields, widths and sizes'), diff --git a/tests/protocols/internet/test_hip_unit.py b/tests/protocols/internet/test_hip_unit.py index 333f57f3c..5826c0996 100644 --- a/tests/protocols/internet/test_hip_unit.py +++ b/tests/protocols/internet/test_hip_unit.py @@ -640,7 +640,7 @@ def test_hip_parameter_readers_cover_simple_models_and_guards(self) -> None: ).suite_id, (HITSuite.ECDSA_SHA_384,)) self.assertEqual(proto._read_param_cert( hip_schema.CertParameter(type=Parameter.CERT, len=8, - cert_group=Group.NIST_P_256, + cert_group=7, cert_count=1, cert_id=2, cert_type=Certificate.X_509_v3, cert=b'cert'), @@ -679,13 +679,21 @@ def test_hip_parameter_readers_cover_simple_models_and_guards(self) -> None: options=options, ).opaque, b'us') - self.assertEqual(proto._read_param_reg_info( + # The octets are kept verbatim; the durations follow + # :rfc:`8003#section-4.1`'s ``2^((lifetime - 64)/8)`` seconds, so the + # ``Max Lifetime`` octet 72 is two seconds (#2281). + reg_info = proto._read_param_reg_info( hip_schema.RegInfoParameter(type=Parameter.REG_INFO, len=3, - min_lifetime=1, max_lifetime=2, + min_lifetime=0, max_lifetime=72, reg_info=[Registration.RENDEZVOUS]), version=2, options=options, - ).lifetime.max, datetime.timedelta(seconds=2)) + ) + self.assertEqual(reg_info.lifetime.max, datetime.timedelta(seconds=2)) + self.assertEqual(reg_info.lifetime.max_exponent, 72) + # Octet 0 is the RFC's own special zero-second value. + self.assertEqual(reg_info.lifetime.min, datetime.timedelta(0)) + self.assertEqual(reg_info.lifetime.min_exponent, 0) self.assertEqual(proto._read_param_reg_request( hip_schema.RegRequestParameter(type=Parameter.REG_REQUEST, len=2, lifetime=3, @@ -693,13 +701,16 @@ def test_hip_parameter_readers_cover_simple_models_and_guards(self) -> None: version=2, options=options, ).reg_type, (Registration.RELAY_UDP_HIP,)) - self.assertEqual(proto._read_param_reg_response( + # Octet 80 is ``2^((80 - 64)/8)`` = 4 seconds (#2281). + reg_response = proto._read_param_reg_response( hip_schema.RegResponseParameter(type=Parameter.REG_RESPONSE, len=2, - lifetime=4, + lifetime=80, reg_response=[Registration.RELAY_UDP_ESP]), version=2, options=options, - ).lifetime, datetime.timedelta(seconds=4)) + ) + self.assertEqual(reg_response.lifetime, datetime.timedelta(seconds=4)) + self.assertEqual(reg_response.lifetime_exponent, 80) reg_failed = proto._read_param_reg_failed( hip_schema.RegFailedParameter(type=Parameter.REG_FAILED, len=2, failure_type=RegistrationFailure.Insufficient_resources, @@ -810,11 +821,12 @@ def test_hip_parameter_readers_cover_simple_models_and_guards(self) -> None: version=2, options=options, ).port, 10502) + # A hop count, not a duration (#2282). self.assertEqual(proto._read_param_overlay_ttl( hip_schema.OverlayTTLParameter(type=Parameter.OVERLAY_TTL, len=4, ttl=30), version=2, options=options, - ).ttl, datetime.timedelta(seconds=30)) + ).ttl, 30) self.assertTrue(proto._read_param_route_via( hip_schema.RouteViaParameter(type=Parameter.ROUTE_VIA, len=20, flags={'symmetric': 1, 'must_follow': 0, 'reserved': 0}, @@ -1116,7 +1128,7 @@ def test_hip_parameter_constructors_cover_keyword_paths(self) -> None: self.assertEqual(proto._make_param_cert( Parameter.CERT, version=2, - cert_group=Group.NIST_P_256, + cert_group=7, cert_count=1, cert_id=2, cert_type=Certificate.X_509_v3, @@ -1271,7 +1283,7 @@ def test_hip_parameter_constructors_cover_keyword_paths(self) -> None: self.assertEqual(proto._make_param_overlay_ttl( Parameter.OVERLAY_TTL, version=2, - ttl=datetime.timedelta(seconds=30), + ttl=30, ).ttl, 30) self.assertTrue(proto._make_param_route_via( Parameter.ROUTE_VIA, @@ -1568,7 +1580,7 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_cert( Parameter.CERT, hip_data.CertParameter(type=Parameter.CERT, critical=False, - length=12, cert_group=Group.NIST_P_256, + length=12, cert_group=7, cert_count=1, cert_id=2, cert_type=Certificate.X_509_v3, cert=b'cert'), @@ -1590,34 +1602,37 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> version=2, ).opaque, b'ers') - lifetime = hip_data.Lifetime(min=dt1, max=dt2) + # The exponent octets on the data model are what a rebuild writes back, + # not the decoded durations beside them (#2281). + lifetime = hip_data.Lifetime(min_exponent=64, max_exponent=72, + min=dt1, max=dt2) self.assertEqual(proto._make_param_reg_info( Parameter.REG_INFO, hip_data.RegInfoParameter(type=Parameter.REG_INFO, critical=False, length=8, lifetime=lifetime, reg_type=(Registration.RENDEZVOUS,)), version=2, - ).max_lifetime, 2) + ).max_lifetime, 72) self.assertEqual(proto._make_param_reg_info(Parameter.REG_INFO, version=2).reg_info, []) self.assertEqual(proto._make_param_reg_request( Parameter.REG_REQUEST, hip_data.RegRequestParameter(type=Parameter.REG_REQUEST, critical=False, length=8, - lifetime=dt1, + lifetime_exponent=64, lifetime=dt1, reg_type=(Registration.RELAY_UDP_HIP,)), version=2, - ).lifetime, 1) + ).lifetime, 64) self.assertEqual(proto._make_param_reg_request(Parameter.REG_REQUEST, version=2).reg_request, []) self.assertEqual(proto._make_param_reg_response( Parameter.REG_RESPONSE, hip_data.RegResponseParameter(type=Parameter.REG_RESPONSE, critical=False, length=8, - lifetime=dt1, + lifetime_exponent=64, lifetime=dt1, reg_type=(Registration.RELAY_UDP_ESP,)), version=2, - ).lifetime, 1) + ).lifetime, 64) self.assertEqual(proto._make_param_reg_response(Parameter.REG_RESPONSE, version=2).reg_response, []) self.assertEqual(proto._make_param_reg_failed( @@ -1692,13 +1707,15 @@ def test_hip_parameter_constructors_cover_data_model_and_default_paths(self) -> self.assertEqual(proto._make_param_transaction_id( Parameter.TRANSACTION_ID, hip_data.TransactionIDParameter(type=Parameter.TRANSACTION_ID, - critical=False, length=8, id=88), + critical=False, length=8, id=88, + id_len=4), version=2, ).id, 88) self.assertEqual(proto._make_param_overlay_id( Parameter.OVERLAY_ID, hip_data.OverlayIDParameter(type=Parameter.OVERLAY_ID, - critical=False, length=8, id=99), + critical=False, length=8, id=99, + id_len=4), version=2, ).id, 99) flags = hip_data.Flags(symmetric=True, must_follow=True, reserved=0) @@ -1783,7 +1800,7 @@ 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, reserved=bytes(2), length=8, ttl=dt1), + critical=False, reserved=bytes(2), length=8, ttl=1), version=2, ).ttl, 1) self.assertTrue(proto._make_param_route_via( diff --git a/tests/protocols/internet/test_hip_wire_semantics_unit.py b/tests/protocols/internet/test_hip_wire_semantics_unit.py new file mode 100644 index 000000000..b8474a259 --- /dev/null +++ b/tests/protocols/internet/test_hip_wire_semantics_unit.py @@ -0,0 +1,382 @@ +# -*- coding: utf-8 -*- +"""HIP parameters carry the units, widths and registries their RFCs define. + +Seven defects in :mod:`pcapkit.protocols.internet.hip`, each a field read or +written against the wrong definition: + +* #2271 -- ``TRANSACTION_ID`` (:rfc:`6078#section-4.4`) and ``OVERLAY_ID`` + (:rfc:`6079#section-6.1`) size their rebuilt ``Length`` from + :meth:`int.bit_length`, which cannot see a leading zero octet, so an + identifier read with ``Length = 4`` was written back as ``Length = 1``. +* #2275 -- ``HIP_TRANSFORM`` is :rfc:`5201` only and its reader says so, but + its maker had no version gate and built octets that reader rejects. +* #2276 -- a ``LOCATOR_SET`` locator built with ``spi=`` left ``type`` at ``0`` + while setting ``Locator Length`` to ``5``, a pair + :func:`~pcapkit.protocols.schema.internet.hip.locator_value_selector` + refuses; :rfc:`8046#section-4.2` makes the SPI form Type 1. +* #2281 -- ``REG_INFO``, ``REG_REQUEST`` and ``REG_RESPONSE`` read their + lifetime octets as counts of seconds, where :rfc:`8003#section-4.1` makes a + non-zero octet an exponent standing for ``2^((lifetime - 64)/8)`` seconds. +* #2282 -- ``OVERLAY_TTL``'s ``ttl`` parsed as a :class:`~datetime.timedelta`, + but :rfc:`6079#section-6.2` gives it as "the number of overlay hops this + packet has left". +* #2283 -- ``CERT``'s group decoded through the Diffie-Hellman group registry, + so a group of ``7`` read back as the curve ``NIST_P_256``; + :rfc:`8002#section-2` leaves that namespace to each sending host. +* #2299 -- ``HOST_ID``'s maker prefixed its errors ``[HIPv2]`` rather than the + ``HIPv{version}: [ParamNo {code}]`` every other HIP message uses. + +Every case builds its own octets in memory and reads no capture. :class:`HIP` +is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, for the reason +:mod:`tests.protocols.link.test_ethernet_mac_roundtrip_unit` gives. + +""" + +import datetime +import importlib.util +import io +import unittest + +from tests._support import reimport_once_per_class + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Test tier (#1958): hand-picked round-trip cases, not a sweep. +__tier__ = 'ALL' + +#: Header fields shared by every HIP packet built here. +HIP_BASE = { + 'next': 6, 'packet': 1, 'checksum': b'\x00\x00', + 'controls_anonymous': False, 'shit': 0, 'rhit': 0, 'payload': b'', +} + +#: Offset of the first parameter: the 8-octet fixed header plus the two HITs. +PARAM_OFFSET = 40 + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class HIPWireSemanticsTests(unittest.TestCase): + """One test per defect, each pinning the RFC's own reading of the field.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + # -- helpers ---------------------------------------------------------- + + def _packet(self, code: 'object', value: 'object', version: 'int' = 2) -> 'bytes': + """Build a whole HIP packet carrying the one parameter ``code``.""" + from pcapkit.protocols.internet.hip import HIP + + return HIP(parameters=[(code, value)], extension=True, + version=version, **HIP_BASE).data # type: ignore[arg-type] + + def _params(self, code: 'object', value: 'object', version: 'int' = 2) -> 'bytes': + """Just the parameter octets of such a packet.""" + return self._packet(code, value, version)[PARAM_OFFSET:] + + def _wrap(self, raw: 'bytes', version: 'int' = 2) -> 'bytes': + """Build a whole HIP packet whose parameter octets are ``raw``.""" + from pcapkit.protocols.internet.hip import HIP + + return HIP(parameters=[raw], extension=True, + version=version, **HIP_BASE).data + + def _parse(self, code: 'object', raw: 'bytes', version: 'int' = 2) -> 'object': + """Parse the parameter octets ``raw`` and return the parsed data.""" + from pcapkit.protocols.internet.hip import HIP + + wire = self._wrap(raw, version) + return HIP(io.BytesIO(wire), len(wire), extension=True).info.parameters[code] # type: ignore[index] + + # -- #2271 ------------------------------------------------------------ + + def test_an_identifier_rebuilds_at_the_width_it_was_read_with(self) -> None: + """#2271: leading zero octets survive a parse-rebuild round trip.""" + from pcapkit.const.hip.parameter import Parameter + + # ``len=4, id=1`` -- three leading zero octets, which ``bit_length`` + # cannot see and the padded ``length`` cannot recover. + for code, raw_hex in ((Parameter.TRANSACTION_ID, '11e4000400000001'), + (Parameter.OVERLAY_ID, '11f0000400000001')): + with self.subTest(code=code): + raw = bytes.fromhex(raw_hex) + parsed = self._parse(code, raw) + + self.assertEqual(parsed.id, 1) # type: ignore[attr-defined] + self.assertEqual(parsed.id_len, 4) # type: ignore[attr-defined] + # The whole point: byte-exact, where it used to come back as + # ``...000101000000``. + self.assertEqual(self._params(code, parsed).hex(), raw_hex) + + def test_an_identifier_width_may_be_declared_or_inferred(self) -> None: + """#2271: ``id_len`` overrides, and ``bit_length`` is the fallback.""" + from pcapkit.const.hip.parameter import Parameter + from pcapkit.utilities.exceptions import ProtocolError + + for code in (Parameter.TRANSACTION_ID, Parameter.OVERLAY_ID): + with self.subTest(code=code): + # Declared width wins, so a from-scratch build can ask for the + # padded form. + self.assertEqual( + self._params(code, {'id': 1, 'id_len': 4})[2:4].hex(), '0004') + # With nothing declaring it, the value's own width is all there + # is -- the documented fallback, not a regression. + self.assertEqual( + self._params(code, {'id': 1})[2:4].hex(), '0001') + # A width too narrow for the value is refused rather than + # silently truncating the identifier. + with self.assertRaisesRegex(ProtocolError, 'invalid identifier width'): + self._params(code, {'id': 0x1234, 'id_len': 1}) + + # -- #2275 ------------------------------------------------------------ + + def test_hip_transform_is_refused_outside_version_1(self) -> None: + """#2275: the *maker* gates on version, as its own reader does. + + The maker is called directly rather than through + :meth:`HIP.make `, because a + whole-packet build raised this same message from + ``_read_param_hip_transform`` all along -- the packet is parsed back on + the way out. Going through the public path therefore cannot tell a + gated maker from an ungated one, which is the distinction the issue is + about: the maker used to return ``0241000200010000`` for HIPv2, octets + its own reader rejects. + + """ + from pcapkit.const.hip.parameter import Parameter + from pcapkit.const.hip.suite import Suite + from pcapkit.protocols.internet.hip import HIP + from pcapkit.utilities.exceptions import ProtocolError + + proto = HIP.__new__(HIP) + suites = [Suite.AES_CBC_with_HMAC_SHA1] + + with self.assertRaisesRegex( + ProtocolError, rf'HIPv2: \[ParamNo {Parameter.HIP_TRANSFORM}\] invalid parameter'): + proto._make_param_hip_transform(Parameter.HIP_TRANSFORM, version=2, suites=suites) + + # HIPv1, where :rfc:`5201#section-5.2.7` does define it, still builds -- + # both through the maker and all the way through a packet. + self.assertEqual( + proto._make_param_hip_transform( + Parameter.HIP_TRANSFORM, version=1, suites=suites).pack().hex(), + '0241000200010000') + self.assertEqual( + self._params(Parameter.HIP_TRANSFORM, {'suites': suites}, version=1).hex(), + '0241000200010000') + + # -- #2276 ------------------------------------------------------------ + + def test_a_locator_with_an_spi_needs_no_explicit_type(self) -> None: + """#2276: ``spi=`` alone builds an :rfc:`8046` Locator Type 1.""" + from pcapkit.const.hip.parameter import Parameter + + derived = self._params(Parameter.LOCATOR_SET, + {'locator_set': [{'ip': '2001:db8::1', 'spi': 7}]}) + restated = self._params(Parameter.LOCATOR_SET, + {'locator_set': [{'ip': '2001:db8::1', + 'type': 1, 'spi': 7}]}) + + # ``spi=`` alone used to raise ``FieldValueError: invalid locator type + # or length``; it now agrees with the form that restates the type. + self.assertEqual(derived.hex(), restated.hex()) + # Locator Type 1, Locator Length 5 -- the SPI plus a 128-bit address. + self.assertEqual(derived[4:7].hex(), '000105') + + def test_a_locator_without_an_spi_is_type_0(self) -> None: + """#2276: the plain-address form is unchanged.""" + from pcapkit.const.hip.parameter import Parameter + + octets = self._params(Parameter.LOCATOR_SET, + {'locator_set': [{'ip': '2001:db8::1'}]}) + self.assertEqual(octets[4:7].hex(), '000004') + + def test_a_locator_type_contradicting_its_spi_is_refused(self) -> None: + """#2276: a stated ``type`` that disagrees with ``spi`` raises.""" + from pcapkit.const.hip.parameter import Parameter + from pcapkit.utilities.exceptions import ProtocolError + + # Type 1 promises an SPI that is not there. + with self.assertRaisesRegex(ProtocolError, 'invalid locator type: 1 does not match spi None'): + self._params(Parameter.LOCATOR_SET, + {'locator_set': [{'ip': '2001:db8::1', 'type': 1}]}) + # Type 0 has no room for the SPI that was given. + with self.assertRaisesRegex(ProtocolError, 'invalid locator type: 0 does not match spi 7'): + self._params(Parameter.LOCATOR_SET, + {'locator_set': [{'ip': '2001:db8::1', 'type': 0, 'spi': 7}]}) + + # -- #2281 ------------------------------------------------------------ + + def test_registration_lifetime_octets_decode_per_rfc_8003(self) -> None: + """#2281: a non-zero octet is an exponent, and ``0`` means zero.""" + from pcapkit.const.hip.parameter import Parameter + + # ``(code, octet, expected)``. The two endpoints are the ones + # :rfc:`8003#section-4.1` names: ``2^(-63/8)`` is "~4 ms" and + # ``2^(191/8)`` is "~178 days". + cases = ( + (0, datetime.timedelta(0)), + (1, datetime.timedelta(seconds=2 ** (-63 / 8))), + (64, datetime.timedelta(seconds=1)), + (72, datetime.timedelta(seconds=2)), + (255, datetime.timedelta(seconds=2 ** (191 / 8))), + ) + + for octet, expected in cases: + with self.subTest(octet=octet): + for code, key in ((Parameter.REG_REQUEST, 'reg_request'), + (Parameter.REG_RESPONSE, 'reg_response')): + raw = self._params(code, {'lifetime_exponent': octet, key: [1]}) + parsed = self._parse(code, raw) + + # The octet read as a count of seconds was the bug: 64 was + # reported as 64 seconds rather than as one second. + self.assertEqual(parsed.lifetime, expected) # type: ignore[attr-defined] + self.assertEqual(parsed.lifetime_exponent, octet) # type: ignore[attr-defined] + + # ``REG_INFO`` carries two of them. + raw = self._params(Parameter.REG_INFO, + {'min_lifetime_exponent': 0, + 'max_lifetime_exponent': 255, 'reg_info': [1]}) + lifetime = self._parse(Parameter.REG_INFO, raw).lifetime # type: ignore[attr-defined] + self.assertEqual(lifetime.min, datetime.timedelta(0)) + self.assertEqual(lifetime.max, datetime.timedelta(seconds=2 ** (191 / 8))) + self.assertEqual((lifetime.min_exponent, lifetime.max_exponent), (0, 255)) + # ~178 days, as the RFC's own prose says. + self.assertEqual(lifetime.max.days, 178) + + def test_a_registration_lifetime_duration_encodes_to_its_octet(self) -> None: + """#2281: the makers take a duration and encode it per the RFC.""" + from pcapkit.const.hip.parameter import Parameter + + # One second is octet 64, not octet 1; the old maker wrote 1. + self.assertEqual( + self._params(Parameter.REG_REQUEST, + {'lifetime': datetime.timedelta(seconds=1), + 'reg_request': [1]})[4], 64) + self.assertEqual( + self._params(Parameter.REG_REQUEST, {'lifetime': 2, 'reg_request': [1]})[4], 72) + # Zero is the RFC's reserved cancellation value rather than an error. + self.assertEqual( + self._params(Parameter.REG_REQUEST, {'lifetime': 0, 'reg_request': [1]})[4], 0) + # REG_INFO's pair, both as durations. + octets = self._params(Parameter.REG_INFO, + {'min_lifetime': 1, 'max_lifetime': 2, 'reg_info': [1]}) + self.assertEqual((octets[4], octets[5]), (64, 72)) + + def test_every_registration_lifetime_octet_round_trips(self) -> None: + """#2281: the raw octet is the rebuild source, so all 256 survive. + + Decoding to a :class:`~datetime.timedelta` and encoding back from it + cannot do this -- most of the 256 octets have no exact duration, which + is why the octet is kept beside it, as ``PUZZLE``'s is (#1323). + + """ + from pcapkit.const.hip.parameter import Parameter + + for code, key in ((Parameter.REG_REQUEST, 'reg_request'), + (Parameter.REG_RESPONSE, 'reg_response')): + for octet in range(256): + with self.subTest(code=code, octet=octet): + raw = self._params(code, {'lifetime_exponent': octet, key: [1]}) + self.assertEqual(raw[4], octet) + parsed = self._parse(code, raw) + self.assertEqual(self._params(code, parsed).hex(), raw.hex()) + + def test_an_out_of_range_registration_lifetime_is_refused(self) -> None: + """#2281: neither a negative duration nor a wide octet is written.""" + from pcapkit.const.hip.parameter import Parameter + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaisesRegex(ProtocolError, 'invalid lifetime'): + self._params(Parameter.REG_REQUEST, + {'lifetime': datetime.timedelta(seconds=-1), 'reg_request': [1]}) + # ``UInt8Field`` wraps rather than raising, so 300 would otherwise be + # written as some other valid-looking duration. + with self.assertRaisesRegex(ProtocolError, 'invalid lifetime'): + self._params(Parameter.REG_REQUEST, + {'lifetime_exponent': 300, 'reg_request': [1]}) + # Past ``2^(191/8)`` seconds the octet no longer fits. + with self.assertRaisesRegex(ProtocolError, 'invalid lifetime'): + self._params(Parameter.REG_REQUEST, + {'lifetime': datetime.timedelta(days=365 * 10), 'reg_request': [1]}) + + # -- #2282 ------------------------------------------------------------ + + def test_overlay_ttl_is_a_hop_count(self) -> None: + """#2282: ``ttl`` is an :obj:`int`, not a duration.""" + from pcapkit.const.hip.parameter import Parameter + + raw = self._params(Parameter.OVERLAY_TTL, {'ttl': 5}) + self.assertEqual(raw.hex(), 'fa0b000400050000') + + parsed = self._parse(Parameter.OVERLAY_TTL, raw) + # It used to come back as ``timedelta(seconds=5)``. + self.assertEqual(parsed.ttl, 5) # type: ignore[attr-defined] + self.assertIsInstance(parsed.ttl, int) # type: ignore[attr-defined] + self.assertNotIsInstance(parsed.ttl, datetime.timedelta) # type: ignore[attr-defined] + self.assertEqual(self._params(Parameter.OVERLAY_TTL, parsed).hex(), raw.hex()) + + # -- #2283 ------------------------------------------------------------ + + def test_cert_group_is_a_plain_octet(self) -> None: + """#2283: no registry stands behind ``CERT group``.""" + from pcapkit.const.hip.parameter import Parameter + + raw = self._params(Parameter.CERT, + {'cert_group': 7, 'cert_count': 1, + 'cert_id': 1, 'cert': b'x'}) + parsed = self._parse(Parameter.CERT, raw) + + # ``7`` used to decode as ````, a Diffie-Hellman + # curve name on a field :rfc:`8002#section-2` leaves to the sender. + self.assertEqual(parsed.cert_group, 7) # type: ignore[attr-defined] + self.assertIs(type(parsed.cert_group), int) # type: ignore[attr-defined] + self.assertEqual(self._params(Parameter.CERT, parsed).hex(), raw.hex()) + + def test_cert_group_defaults_to_zero(self) -> None: + """#2283: the default is neutral, not a curve name.""" + from pcapkit.const.hip.parameter import Parameter + + self.assertEqual(self._params(Parameter.CERT, {'cert': b'x'})[4], 0) + + def test_cert_group_takes_no_enum_lookup_arguments(self) -> None: + """#2283: the namespace arguments existed only for the wrong registry.""" + import inspect + + from pcapkit.protocols.internet.hip import HIP + + keywords = inspect.signature(HIP._make_param_cert).parameters + for dropped in ('cert_group_default', 'cert_group_namespace', + 'cert_group_reversed'): + self.assertNotIn(dropped, keywords) + # ``cert_type`` keeps its own, since :rfc:`8002` really does register it. + self.assertIn('cert_type_namespace', keywords) + + # -- #2299 ------------------------------------------------------------ + + def test_host_id_errors_use_the_house_prefix(self) -> None: + """#2299: ``HIPv2: [ParamNo 705] ...``, not ``[HIPv2] ...``.""" + from pcapkit.const.hip.parameter import Parameter + from pcapkit.protocols.data.internet.hip import HostIdentity + from pcapkit.protocols.internet.hip import HIP + from pcapkit.utilities.exceptions import ProtocolError + + proto = HIP.__new__(HIP) + expected = rf"^HIPv2: \[ParamNo {Parameter.HOST_ID}\] invalid curve: 'bogus'$" + + # The ``hi_curve`` branch. + with self.assertRaisesRegex(ProtocolError, expected): + proto._make_param_host_id(Parameter.HOST_ID, hi_curve='bogus', # type: ignore[arg-type] + hi_pub_key=b'', version=2) + # The branch for a ``HostIdentity`` passed as ``hi``. + with self.assertRaisesRegex(ProtocolError, expected): + proto._make_param_host_id(Parameter.HOST_ID, + hi=HostIdentity(curve='bogus', pubkey=b''), # type: ignore[arg-type] + version=2) + + +if __name__ == '__main__': + unittest.main()