Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 18 additions & 5 deletions pcapkit/protocols/data/internet/hip.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,9 +66,12 @@ class Control(Data):

#: Anonymous flag.
anonymous: 'bool'
#: Reserved bits -- the 15 high-order bits of ``Controls``, carried
#: verbatim so that re-serialising reproduces them.
reserved: 'int'

if TYPE_CHECKING:
def __init__(self, anonymous: 'bool') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,multiple-statements
def __init__(self, anonymous: 'bool', reserved: 'int') -> 'None': ... # pylint: disable=super-init-not-called,unused-argument,multiple-statements


class Parameter(Data):
Expand Down Expand Up @@ -125,6 +128,8 @@ def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', cont
class ESPInfoParameter(Parameter):
"""Data model for HIP ``ESP_INFO`` parameter."""

#: Reserved octets, carried verbatim so that re-serialising reproduces them.
reserved: 'int'
#: KEYMAT index.
index: 'int'
#: Old SDI.
Expand All @@ -133,7 +138,7 @@ class ESPInfoParameter(Parameter):
new_spi: 'int'

if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', index: 'int', old_spi: 'int', new_spi: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long
def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', reserved: 'int', index: 'int', old_spi: 'int', new_spi: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long


@info_final
Expand Down Expand Up @@ -198,8 +203,16 @@ class PuzzleParameter(Parameter):

#: Numeric index.
index: 'int'
#: Lifetime.
lifetime: 'timedelta'
#: The ``Lifetime`` octet as it appears on the wire. :rfc:`7401#section-5.2.4`
#: defines the lifetime as ``2^(value - 32)`` seconds, and this is ``value``.
#: It is what re-serialising writes back, since most of the 256 octets have
#: no exact :class:`~datetime.timedelta`.
lifetime_exponent: 'int'
#: Lifetime, derived from :attr:`lifetime_exponent` to
#: :class:`~datetime.timedelta`'s microsecond resolution, so octets below
#: ``12`` read as ``timedelta(0)``. :obj:`None` for octets of ``79`` and above,
#: whose duration exceeds :attr:`datetime.timedelta.max`.
lifetime: 'Optional[timedelta]'
#: Puzzle data.
opaque: 'bytes'
#: Random number.
Expand All @@ -213,7 +226,7 @@ class PuzzleParameter(Parameter):
rhash_len: 'int'

if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', critical: 'bool', length: 'int', index: 'int', lifetime: 'timedelta', opaque: 'bytes', random: 'int', rhash_len: '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', index: 'int', lifetime_exponent: 'int', lifetime: 'Optional[timedelta]', opaque: 'bytes', random: 'int', rhash_len: 'int') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,redefined-builtin,super-init-not-called,line-too-long


@info_final
Expand Down
73 changes: 61 additions & 12 deletions pcapkit/protocols/internet/hip.py
Original file line number Diff line number Diff line change
Expand Up @@ -583,6 +583,7 @@ def read(self, length: 'Optional[int]' = None, *, extension: bool = False, **kwa
chksum=schema.checksum,
control=Data_Control(
anonymous=bool(schema.control['anonymous']),
reserved=schema.control['reserved'],
),
shit=schema.shit,
rhit=schema.rhit,
Expand Down Expand Up @@ -615,6 +616,7 @@ def make(self,
version: 'int' = 2,
checksum: 'bytes' = b'\x00\x00',
controls_anonymous: 'bool' = False,
controls_reserved: 'int' = 0,
shit: 'int' = 0,
rhit: 'int' = 0,
parameters: 'Optional[list[Schema_Parameter | tuple[Enum_Parameter, dict[str, Any]] | bytes] | Parameter]' = None, # pylint: disable=line-too-long
Expand All @@ -634,6 +636,7 @@ def make(self,
version: HIP version.
checksum: Checksum.
controls_anonymous: If the sender is anonymous.
controls_reserved: The 15 reserved high-order bits of ``Controls``.
shit: Sender's host identity tag (HIT).
rhit: Receiver's host identity tag (HIT).
parameters: HIP parameters.
Expand All @@ -651,7 +654,9 @@ def make(self,

if parameters is not None:
parameters_value, total_length = self._make_hip_param(parameters, version=version)
length = total_length // 8 + 4
# Ceiling division, so that Header Length never declares fewer
# octets than are packed.
length = -(-total_length // 8) + 4
else:
# NOTE: Header Length excludes the first 8 octets but always
# counts the two 16-octet HITs, hence the minimum of 4.
Expand All @@ -670,6 +675,7 @@ def make(self,
},
checksum=checksum,
control = {
'reserved': controls_reserved,
'anonymous': controls_anonymous,
},
shit=shit,
Expand Down Expand Up @@ -759,9 +765,11 @@ def _make_data(cls, data: 'Data_HIP') -> 'dict[str, Any]': # type: ignore[overr
'version': data.version,
'checksum': data.chksum,
'controls_anonymous': data.control.anonymous,
'controls_reserved': data.control.reserved,
'shit': data.shit,
'rhit': data.rhit,
'parameters': data.parameters,
# ``read`` sets ``parameters`` only when the packet carries some.
'parameters': getattr(data, 'parameters', None),
'payload': cls._make_payload(data),
}

Expand Down Expand Up @@ -878,6 +886,7 @@ def _read_param_esp_info(self, schema: 'Schema_ESPInfoParameter', *, version: 'i
type=schema.type,
critical=bool(schema.type & 0b1),
length=parameter_total_len(schema.len),
reserved=schema.reserved,
index=schema.index,
old_spi=schema.old_spi,
new_spi=schema.new_spi,
Expand Down Expand Up @@ -1080,7 +1089,8 @@ def _read_param_puzzle(self, schema: 'Schema_PuzzleParameter', *, version: 'int'
critical=bool(schema.type & 0b1),
length=parameter_total_len(schema.len),
index=_numk,
lifetime=datetime.timedelta(seconds=2 ** (_time - 32)),
lifetime_exponent=_time,
lifetime=self._read_puzzle_lifetime(_time),
opaque=_opak,
random=_rand,
# Keep the field's on-wire width, which ``_rand`` alone cannot carry:
Expand All @@ -1091,6 +1101,28 @@ def _read_param_puzzle(self, schema: 'Schema_PuzzleParameter', *, version: 'int'
)
return puzzle

@staticmethod
def _read_puzzle_lifetime(octet: 'int') -> 'Optional[timedelta]':
"""Decode a ``PUZZLE`` ``Lifetime`` octet into a duration.

:rfc:`7401#section-5.2.4` gives the lifetime as ``2^(value - 32)``
seconds. The result is informational only: :meth:`_make_param_puzzle`
writes back the octet itself, since :class:`~datetime.timedelta` rounds
octets below ``12`` to zero and cannot hold those of ``79`` and above.

Args:
octet: the ``Lifetime`` octet

Returns:
The lifetime, or :obj:`None` if it exceeds
:attr:`datetime.timedelta.max`.

"""
try:
return datetime.timedelta(seconds=2 ** (octet - 32))
except OverflowError:
return None

def _read_param_solution(self, schema: 'Schema_SolutionParameter', *, version: 'int',
options: 'Parameter') -> 'Data_SolutionParameter': # pylint: disable=unused-argument
"""Read HIP ``SOLUTION`` parameter.
Expand Down Expand Up @@ -2966,6 +2998,10 @@ def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parame
parameters_list = [] # type: list[Schema_Parameter | bytes]
for schema in parameters:
if isinstance(schema, bytes):
# :rfc:`7401#section-5.2.1` pads every parameter to a
# multiple of 8 octets, and Header Length counts in those
# units; a raw parameter is taken to omit its padding.
schema += bytes(-len(schema) % 8)
parameters_list.append(schema)
total_length += len(schema)
elif isinstance(schema, Schema):
Expand Down Expand Up @@ -3035,6 +3071,7 @@ def _make_param_unassigned(self, code: 'Enum_Parameter', param: 'Optional[Data_U

def _make_param_esp_info(self, code: 'Enum_Parameter', param: 'Optional[Data_ESPInfoParameter]' = None, *, # pylint: disable=unused-argument
version: 'int',
reserved: 'int' = 0,
index: 'int' = 0,
old_spi: 'int' = 0,
new_spi: 'int' = 0,
Expand All @@ -3045,6 +3082,7 @@ def _make_param_esp_info(self, code: 'Enum_Parameter', param: 'Optional[Data_ESP
code: parameter code
param: parameter data
version: HIP protocol version
reserved: reserved octets, zero when sent (:rfc:`7402#section-5.1.1`)
index: KEYMAT index
old_spi: old SPI
new_spi: new SPI
Expand All @@ -3055,13 +3093,15 @@ def _make_param_esp_info(self, code: 'Enum_Parameter', param: 'Optional[Data_ESP

"""
if param is not None:
reserved = param.reserved
index = param.index
old_spi = param.old_spi
new_spi = param.new_spi

return Schema_ESPInfoParameter(
type=code,
len=12,
reserved=reserved,
index=index,
old_spi=old_spi,
new_spi=new_spi,
Expand Down Expand Up @@ -3248,10 +3288,7 @@ def _make_puzzle_lifetime(code: 'Enum_Parameter', version: 'int',
# ``math.log2`` raises a bare :exc:`ValueError` at zero and below. That is
# not a :class:`~pcapkit.utilities.exceptions.BaseError`, so it would
# escape the library's error handling with a message naming neither HIP
# nor the field. Conformant input reaches it: a ``Lifetime`` octet of
# ``0x00`` means ``2^-32`` seconds, below
# :class:`~datetime.timedelta`'s microsecond resolution, so parsing one
# yields ``timedelta(0)`` and re-serialising it lands here.
# nor the field.
if seconds <= 0:
raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid lifetime: '
f'{seconds} is not a positive number of seconds')
Expand Down Expand Up @@ -3347,6 +3384,7 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl
version: 'int',
index: 'int' = 0,
lifetime: 'timedelta | int' = 0,
lifetime_exponent: 'Optional[int]' = None,
opaque: 'bytes' = b'',
random: 'int' = 0,
rhash_len: 'Optional[int]' = None,
Expand All @@ -3358,7 +3396,10 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl
param: parameter data
version: HIP protocol version
index: #K index
lifetime: lifetime
lifetime: lifetime, encoded to the nearest lower ``Lifetime`` octet;
ignored when ``lifetime_exponent`` is given
lifetime_exponent: the ``Lifetime`` octet itself, ``value`` in
:rfc:`7401#section-5.2.4`'s ``2^(value - 32)`` seconds
opaque: opaque data
random: random #I value
rhash_len: on-wire width of ``Random #I``, in bits; defaults to the
Expand All @@ -3369,16 +3410,24 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl
Returns:
HIP parameter schema.

Raises:
ProtocolError: If ``lifetime_exponent`` does not fit one octet, or
``lifetime`` cannot be encoded; see :meth:`_make_puzzle_lifetime`.

"""
if param is not None:
index = param.index
lifetime = self._make_puzzle_lifetime(code, version, param.lifetime)
lifetime_exponent = param.lifetime_exponent
opaque = param.opaque
random = param.random
if rhash_len is None:
rhash_len = param.rhash_len
else:
lifetime = self._make_puzzle_lifetime(code, version, lifetime)

if lifetime_exponent is None:
lifetime_exponent = self._make_puzzle_lifetime(code, version, lifetime)
elif not 0 <= lifetime_exponent <= 0xFF:
raise ProtocolError(f'HIPv{version}: [ParamNo {code}] invalid lifetime: '
f'{lifetime_exponent} is outside the one-octet Lifetime field')

return Schema_PuzzleParameter(
type=code,
Expand All @@ -3387,7 +3436,7 @@ def _make_param_puzzle(self, code: 'Enum_Parameter', param: 'Optional[Data_Puzzl
# spells the same quantity ``4 + RHASH_len / 8``.
len=4 + self._make_puzzle_field_width(code, version, rhash_len, random),
index=index,
lifetime=lifetime,
lifetime=lifetime_exponent,
opaque=opaque,
random=random,
)
Expand Down
11 changes: 8 additions & 3 deletions pcapkit/protocols/schema/internet/hip.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,8 @@ class ControlsType(TypedDict):

#: Anonymous.
anonymous: bool
#: Reserved bits.
reserved: int

class LocatorFlags(TypedDict):
"""Locator flags."""
Expand Down Expand Up @@ -517,8 +519,10 @@ def __init__(self, type: 'Enum_Parameter', len: 'int', value: 'bytes') -> 'None'
class ESPInfoParameter(Parameter, code=Enum_Parameter.ESP_INFO):
"""Header schema for HIP ``ESP_INFO`` parameters."""

#: Reserved.
reserved: 'bytes' = PaddingField(length=2)
#: Reserved, "zero when sent, ignored when received" (:rfc:`7402#section-5.1.1`).
#: Read as a value rather than as padding, so that re-serialising a parsed
#: parameter reproduces the octets it arrived with.
reserved: 'int' = UInt16Field()
#: Key management index.
index: 'int' = UInt16Field()
#: Old SPI.
Expand All @@ -529,7 +533,7 @@ class ESPInfoParameter(Parameter, code=Enum_Parameter.ESP_INFO):
padding: 'bytes' = PaddingField(length=parameter_padding_len)

if TYPE_CHECKING:
def __init__(self, type: 'Enum_Parameter', len: 'int', index: 'int',
def __init__(self, type: 'Enum_Parameter', len: 'int', reserved: 'int', index: 'int',
old_spi: 'int', new_spi: 'int') -> 'None': ...


Expand Down Expand Up @@ -1567,6 +1571,7 @@ class HIP(Schema):
control: 'ControlsType' = BitField(
length=2,
namespace={
'reserved': (0, 15),
'anonymous': (15, 1),
},
)
Expand Down
Loading
Loading