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
298 changes: 149 additions & 149 deletions pcapkit/protocols/internet/esp.py

Large diffs are not rendered by default.

148 changes: 74 additions & 74 deletions pcapkit/protocols/internet/hip.py
Original file line number Diff line number Diff line change
Expand Up @@ -886,6 +886,80 @@ def _read_hip_param(self, length: 'int', *, version: 'int') -> 'Parameter': # p
raise ProtocolError(f'HIPv{version}: invalid format')
return options

def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parameter, Data_Parameter | dict[str, Any]] | bytes] | Parameter', *, # pylint: disable=line-too-long
version: 'int') -> 'tuple[list[Schema_Parameter | bytes], int]':
"""Make HIP parameter.

A ``(code, value)`` pair carries either a mapping of keyword arguments
for the parameter's constructor, or a parsed
:class:`~pcapkit.protocols.data.internet.hip.Parameter` data model,
which goes to the constructor positionally through
:meth:`_make_hip_param_data` rather than expanded as keyword arguments
(:issue:`2352`).

Args:
parameters: HIP parameters
version: HIP protocol version

Returns:
HIP parameters and total length.

"""
total_length = 0
if isinstance(parameters, list):
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):
schema_packed = schema.pack()

parameters_list.append(schema_packed)
total_length += len(schema_packed)
else:
code, args = cast('tuple[Enum_Parameter, Data_Parameter | dict[str, Any]]',
schema)
name = self._lookup_registry(self.__parameter__, code)
if isinstance(name, str):
meth_name = f'_make_param_{name}'
meth = cast('ParameterConstructor',
getattr(self, meth_name, self._make_param_unassigned))
else:
meth = name[1]

if isinstance(args, Data_Parameter):
data = self._make_hip_param_data(meth, code, args,
version=version) # type: Schema_Parameter
else:
data = meth(code, version=version, **args)
data_packed = data.pack()

parameters_list.append(data)
total_length += len(data_packed)
return parameters_list, total_length

parameters_list = []
for code, param in parameters.items(multi=True):
name = self._lookup_registry(self.__parameter__, code)
if isinstance(name, str):
meth_name = f'_make_param_{name}'
meth = cast('ParameterConstructor',
getattr(self, meth_name, self._make_param_unassigned))
else:
meth = name[1]

data = self._make_hip_param_data(meth, code, param, version=version)
data_packed = data.pack()

parameters_list.append(data)
total_length += len(data_packed)
return parameters_list, total_length

def _read_param_unassigned(self, schema: 'Schema_UnassignedParameter', *, version: 'int', # pylint: disable=unused-argument
options: 'Parameter') -> 'Data_UnassignedParameter': # pylint: disable=unused-argument
"""Read HIP unassigned parameters.
Expand Down Expand Up @@ -3199,80 +3273,6 @@ def _read_param_relay_hmac(self, schema: 'Schema_RelayHMACParameter', *, version
)
return relay_hmac

def _make_hip_param(self, parameters: 'list[Schema_Parameter | tuple[Enum_Parameter, Data_Parameter | dict[str, Any]] | bytes] | Parameter', *, # pylint: disable=line-too-long
version: 'int') -> 'tuple[list[Schema_Parameter | bytes], int]':
"""Make HIP parameter.

A ``(code, value)`` pair carries either a mapping of keyword arguments
for the parameter's constructor, or a parsed
:class:`~pcapkit.protocols.data.internet.hip.Parameter` data model,
which goes to the constructor positionally through
:meth:`_make_hip_param_data` rather than expanded as keyword arguments
(:issue:`2352`).

Args:
parameters: HIP parameters
version: HIP protocol version

Returns:
HIP parameters and total length.

"""
total_length = 0
if isinstance(parameters, list):
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):
schema_packed = schema.pack()

parameters_list.append(schema_packed)
total_length += len(schema_packed)
else:
code, args = cast('tuple[Enum_Parameter, Data_Parameter | dict[str, Any]]',
schema)
name = self._lookup_registry(self.__parameter__, code)
if isinstance(name, str):
meth_name = f'_make_param_{name}'
meth = cast('ParameterConstructor',
getattr(self, meth_name, self._make_param_unassigned))
else:
meth = name[1]

if isinstance(args, Data_Parameter):
data = self._make_hip_param_data(meth, code, args,
version=version) # type: Schema_Parameter
else:
data = meth(code, version=version, **args)
data_packed = data.pack()

parameters_list.append(data)
total_length += len(data_packed)
return parameters_list, total_length

parameters_list = []
for code, param in parameters.items(multi=True):
name = self._lookup_registry(self.__parameter__, code)
if isinstance(name, str):
meth_name = f'_make_param_{name}'
meth = cast('ParameterConstructor',
getattr(self, meth_name, self._make_param_unassigned))
else:
meth = name[1]

data = self._make_hip_param_data(meth, code, param, version=version)
data_packed = data.pack()

parameters_list.append(data)
total_length += len(data_packed)
return parameters_list, total_length

def _make_hip_param_data(self, meth: 'ParameterConstructor', code: 'Enum_Parameter',
param: 'Data_Parameter', *,
version: 'int') -> 'Schema_Parameter':
Expand Down
196 changes: 98 additions & 98 deletions pcapkit/protocols/internet/hopopt.py
Original file line number Diff line number Diff line change
Expand Up @@ -596,6 +596,104 @@ def _read_hopopt_options(self, length: 'int') -> 'Option':
raise ProtocolError('HOPOPT: invalid format')
return options

def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option') -> 'tuple[list[Schema_Option | bytes], int]':
"""Make options for HOPOPT.

Args:
options: HOPOPT options

Returns:
Tuple of options and total length of options.

Note:
Options given as a :obj:`list` are fresh input: padding options in
it are dropped, and padding is inserted after each option to align
the header. Options given as a mapping, such as the
:class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` of parsed
data, are emitted in order with their own padding options kept, and
padding is added only at the end if the header is not aligned.

The returned length is that of the options area alone, the value
:meth:`make` derives the ``Hdr Ext Len`` field from. Since the two
octets of the fixed header precede it, an options area is well
formed only when it is 2 octets short of a multiple of 8 --
including when it holds no real options at all, in which case it is
6 octets of pure padding rather than empty.

"""
total_length = 0
if isinstance(options, list):
options_list = [] # type: list[Schema_Option | bytes]
for schema in options:
if isinstance(schema, bytes):
code = Enum_Option.get(schema[0])
if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default
continue

opt = schema # type: bytes | Schema_Option
opt_len = len(schema)
elif isinstance(schema, Schema):
code = schema.type
if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default
continue

opt = schema
opt_len = len(schema.pack())
else:
code, args = cast('tuple[Enum_Option, dict[str, Any]]', schema)
if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default
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_none))
else:
meth = name[1]

opt = meth(code, **args)
opt_len = len(opt.pack())

options_list.append(opt)
total_length += opt_len

# force alignment of the header (fixed part included) to 8 octets
pad_opts, pad_len = self._make_pad_options(total_length + 2)
options_list.extend(pad_opts)
total_length += pad_len

# an options area holding nothing at all is still 6 octets long
pad_opts, pad_len = self._make_pad_options(total_length + 2)
options_list.extend(pad_opts)
total_length += pad_len
return options_list, total_length

# NOTE: Options given as parsed data are emitted as they are, padding
# included and in place, so that a parsed header rebuilds byte for
# byte; only a header left short of alignment gets trailing padding.
options_list = []
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_none))
else:
meth = name[1]

opt = meth(code, option)
opt_len = len(opt.pack())

options_list.append(opt)
total_length += opt_len

# an options area holding nothing at all is still 6 octets long
pad_opts, pad_len = self._make_pad_options(total_length + 2)
options_list.extend(pad_opts)
total_length += pad_len
return options_list, total_length

def _read_opt_none(self, schema: 'Schema_UnassignedOption', *, options: 'Option') -> 'Data_UnassignedOption': # pylint: disable=unused-argument
"""Read HOPOPT unassigned options.

Expand Down Expand Up @@ -1389,104 +1487,6 @@ def _make_pad_options(self, offset: 'int') -> 'tuple[list[Schema_PadOption], int
return [self._make_opt_pad(Enum_Option.Pad1, length=0)], pad_len # type: ignore[arg-type]
return [self._make_opt_pad(Enum_Option.PadN, length=pad_len - 2)], pad_len # type: ignore[arg-type]

def _make_hopopt_options(self, options: 'list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes] | Option') -> 'tuple[list[Schema_Option | bytes], int]':
"""Make options for HOPOPT.

Args:
options: HOPOPT options

Returns:
Tuple of options and total length of options.

Note:
Options given as a :obj:`list` are fresh input: padding options in
it are dropped, and padding is inserted after each option to align
the header. Options given as a mapping, such as the
:class:`~pcapkit.corekit.infoclass.OrderedMultiInfo` of parsed
data, are emitted in order with their own padding options kept, and
padding is added only at the end if the header is not aligned.

The returned length is that of the options area alone, the value
:meth:`make` derives the ``Hdr Ext Len`` field from. Since the two
octets of the fixed header precede it, an options area is well
formed only when it is 2 octets short of a multiple of 8 --
including when it holds no real options at all, in which case it is
6 octets of pure padding rather than empty.

"""
total_length = 0
if isinstance(options, list):
options_list = [] # type: list[Schema_Option | bytes]
for schema in options:
if isinstance(schema, bytes):
code = Enum_Option.get(schema[0])
if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default
continue

opt = schema # type: bytes | Schema_Option
opt_len = len(schema)
elif isinstance(schema, Schema):
code = schema.type
if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default
continue

opt = schema
opt_len = len(schema.pack())
else:
code, args = cast('tuple[Enum_Option, dict[str, Any]]', schema)
if code in (Enum_Option.Pad1, Enum_Option.PadN): # ignore padding options by default
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_none))
else:
meth = name[1]

opt = meth(code, **args)
opt_len = len(opt.pack())

options_list.append(opt)
total_length += opt_len

# force alignment of the header (fixed part included) to 8 octets
pad_opts, pad_len = self._make_pad_options(total_length + 2)
options_list.extend(pad_opts)
total_length += pad_len

# an options area holding nothing at all is still 6 octets long
pad_opts, pad_len = self._make_pad_options(total_length + 2)
options_list.extend(pad_opts)
total_length += pad_len
return options_list, total_length

# NOTE: Options given as parsed data are emitted as they are, padding
# included and in place, so that a parsed header rebuilds byte for
# byte; only a header left short of alignment gets trailing padding.
options_list = []
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_none))
else:
meth = name[1]

opt = meth(code, option)
opt_len = len(opt.pack())

options_list.append(opt)
total_length += opt_len

# an options area holding nothing at all is still 6 octets long
pad_opts, pad_len = self._make_pad_options(total_length + 2)
options_list.extend(pad_opts)
total_length += pad_len
return options_list, total_length

def _make_opt_none(self, code: 'Enum_Option', opt: 'Optional[Data_UnassignedOption]' = None, *,
data: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_UnassignedOption':
Expand Down
2 changes: 1 addition & 1 deletion pcapkit/protocols/internet/ipsec.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ class IPsec(Internet[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract-
"""

##########################################################################
# Data models.
# Methods.
##########################################################################

@classmethod
Expand Down
Loading
Loading