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
46 changes: 32 additions & 14 deletions pcapkit/protocols/internet/hopopt.py
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,17 @@ class HOPOPT(IPv6_Ext[Data_HOPOPT, Schema_HOPOPT],
* A PadN option's data and the "Not Used" octet of a Quick-Start report
are kept as captured rather than zeroed (:issue:`1223`).

.. versionchanged:: 1.6.0

* A Quick-Start rate the four-bit rate code cannot carry, or a TTL,
nonce or reserved value outside its field, raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than a bare
:exc:`ValueError`, an :exc:`OverflowError` or a
:exc:`~pcapkit.utilities.exceptions.FieldValueError`, or a different
value written in its place (:issue:`2617`).
* The Quick-Start maker builds an unassigned function code as an
unassigned option, the form the reader keeps one in (:issue:`2617`).

"""

##########################################################################
Expand Down Expand Up @@ -1101,29 +1112,27 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option')
if func == Enum_QSFunction.Quick_Start_Request:
schema_req = cast('Schema_QuickStartRequestOption', schema)

rate = schema_req.flags['rate']
opt = Data_QuickStartRequestOption(
type=schema.type,
action=Enum_OptionAction.get(schema.type >> 6),
change=bool(schema.type & 0b00100000),
length=self._hopopt_option_length(schema_req.len),
func=func,
rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0,
rate=self._read_qs_rate(schema_req.flags['rate']),
ttl=datetime.timedelta(seconds=schema_req.ttl),
nonce=schema_req.nonce['nonce'],
reserved=schema_req.nonce['reserved'],
) # type: Data_QuickStartOption
elif func == Enum_QSFunction.Report_of_Approved_Rate:
schema_rep = cast('Schema_QuickStartReportOption', schema)

rate = schema_rep.flags['rate']
opt = Data_QuickStartReportOption(
type=schema.type,
action=Enum_OptionAction.get(schema.type >> 6),
change=bool(schema.type & 0b00100000),
length=self._hopopt_option_length(schema_rep.len),
func=func,
rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0,
rate=self._read_qs_rate(schema_rep.flags['rate']),
nonce=schema_rep.nonce['nonce'],
reserved=schema_rep.nonce['reserved'],
unused=schema_rep.reserved,
Expand Down Expand Up @@ -1909,20 +1918,24 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption
func_default: default value for QS function type
func_namespace: namespace for QS function type
func_reversed: reversed flag for QS function type
rate: rate (in kbps)
rate: rate (in kbps), rounded down to the nearest rate the
four-bit rate code carries (:rfc:`4782#section-3.1`)
ttl: time to live (in seconds)
nonce: nonce value
reserved: reserved bits (``R``) after the nonce
unused: the "Not Used" octet of a report of approved rate
**kwargs: arbitrary keyword arguments

Returns:
Constructed option schema; an unassigned option schema if ``opt``
is a Quick-Start option parsed as unassigned, see :meth:`_read_opt_qs`.
Constructed option schema. An unassigned option schema if ``opt``
is a Quick-Start option parsed as unassigned, see :meth:`_read_opt_qs`,
or if ``func`` is neither of the assigned functions, in which case
the option data takes the layout of a Quick-Start Request.

Raises:
ProtocolError: If the QS function is neither a Quick-Start Request
nor a Report of Approved Rate.
ProtocolError: If ``func`` does not fit in four bits, the rate code
cannot carry ``rate``, or ``ttl``, ``nonce`` or ``reserved`` does
not fit its field.

"""
if isinstance(opt, Data_UnassignedOption):
Expand All @@ -1937,19 +1950,20 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption
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
rate_val = self._make_qs_rate(rate)
ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds())
nonce = self._make_uint(nonce, bits=30, name='Quick-Start nonce')
reserved = self._make_uint(reserved, bits=2, name='Quick-Start reserved')

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=code,
len=6,
flags={
'func': func_enum,
'rate': rate_val,
},
ttl=ttl_value,
ttl=self._make_uint(ttl_value, bits=8, name='Quick-Start TTL'),
nonce={
'nonce': nonce,
'reserved': reserved,
Expand All @@ -1969,7 +1983,11 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption
'reserved': reserved,
},
)
raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid QS function: {func_enum}')
if not 0 <= func_enum < 16:
raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid QS function: {func_enum}')
ttl_value = self._make_uint(ttl_value, bits=8, name='Quick-Start TTL')
data = bytes([(func_enum << 4) | rate_val, ttl_value]) + ((nonce << 2) | reserved).to_bytes(4, 'big')
return self._make_opt_none(code, data=data)

def _make_opt_rpl(self, code: 'Enum_Option', opt: 'Optional[Data_RPLOption]' = None, *,
down: 'bool' = False,
Expand Down
32 changes: 21 additions & 11 deletions pcapkit/protocols/internet/ipv4.py
Original file line number Diff line number Diff line change
Expand Up @@ -212,9 +212,15 @@ class IPv4(IP[Data_IPv4, Schema_IPv4],

.. versionchanged:: 1.6.0

The header checksum is computed when :meth:`make` is given none, and
verified on read under :data:`~pcapkit.utilities.config.CHECKSUM_VERIFY`
(:issue:`2472`).
* The header checksum is computed when :meth:`make` is given none, and
verified on read under :data:`~pcapkit.utilities.config.CHECKSUM_VERIFY`
(:issue:`2472`).
* A Quick-Start rate the four-bit rate code cannot carry, or a TTL,
nonce or reserved value outside its field, raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than a bare
:exc:`ValueError`, an :exc:`OverflowError` or a
:exc:`~pcapkit.utilities.exceptions.FieldValueError`, or a different
value written in its place (:issue:`2617`).

"""

Expand Down Expand Up @@ -1517,27 +1523,25 @@ def _read_opt_qs(self, schema: 'Schema_QSOption', *, options: 'Option') -> 'Data
if func == Enum_QSFunction.Quick_Start_Request:
schema_req = cast('Schema_QuickStartRequestOption', schema)

rate = schema_req.flags['rate']
opt = Data_QuickStartRequestOption(
code=schema.type,
type=self._read_ipv4_opt_type(schema.type),
length=schema_req.length,
func=func,
rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0,
rate=self._read_qs_rate(schema_req.flags['rate']),
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)

rate = schema_rep.flags['rate']
opt = Data_QuickStartReportOption(
code=schema.type,
type=self._read_ipv4_opt_type(schema.type),
length=schema_rep.length,
func=func,
rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0,
rate=self._read_qs_rate(schema_rep.flags['rate']),
unused=schema_rep.reserved,
nonce=schema_rep.nonce['nonce'],
reserved=schema_rep.nonce['reserved'],
Expand Down Expand Up @@ -2170,7 +2174,8 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt
func_default: default value for QS function type
func_namespace: namespace for QS function type
func_reversed: reversed flag for QS function type
rate: rate (in kbps)
rate: rate (in kbps), rounded down to the nearest rate the
four-bit rate code carries (:rfc:`4782#section-3.1`)
ttl: time to live (in seconds)
unused: the not-used fourth octet of a Report of Approved Rate
nonce: nonce value
Expand All @@ -2184,7 +2189,9 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt
the option takes the eight-octet layout of a Quick-Start Request.

Raises:
ProtocolError: If ``func`` does not fit in four bits.
ProtocolError: If ``func`` does not fit in four bits, the rate code
cannot carry ``rate``, or ``ttl``, ``nonce`` or ``reserved`` does
not fit its field.

"""
if isinstance(option, Data_UnassignedOption):
Expand All @@ -2199,8 +2206,10 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt
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
rate_val = self._make_qs_rate(rate)
ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds())
nonce = self._make_uint(nonce, bits=30, name='Quick-Start nonce')
reserved = self._make_uint(reserved, bits=2, name='Quick-Start reserved')

if func_enum == Enum_QSFunction.Quick_Start_Request:
return Schema_QuickStartRequestOption(
Expand All @@ -2210,7 +2219,7 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt
'func': func_enum,
'rate': rate_val,
},
ttl=ttl_value,
ttl=self._make_uint(ttl_value, bits=8, name='Quick-Start TTL'),
nonce={
'nonce': nonce,
'reserved': reserved,
Expand All @@ -2232,5 +2241,6 @@ def _make_opt_qs(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_QuickSt
)
if not 0 <= func_enum < 16:
raise ProtocolError(f'{self.alias}: [OptNo {kind}] invalid QS function: {func_enum}')
ttl_value = self._make_uint(ttl_value, bits=8, name='Quick-Start TTL')
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))
46 changes: 32 additions & 14 deletions pcapkit/protocols/internet/ipv6_opts.py
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,17 @@ class IPv6_Opts(IPv6_Ext[Data_IPv6_Opts, Schema_IPv6_Opts],
* A PadN option's data and the "Not Used" octet of a Quick-Start report
are kept as captured rather than zeroed (:issue:`1223`).

.. versionchanged:: 1.6.0

* A Quick-Start rate the four-bit rate code cannot carry, or a TTL,
nonce or reserved value outside its field, raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than a bare
:exc:`ValueError`, an :exc:`OverflowError` or a
:exc:`~pcapkit.utilities.exceptions.FieldValueError`, or a different
value written in its place (:issue:`2617`).
* The Quick-Start maker builds an unassigned function code as an
unassigned option, the form the reader keeps one in (:issue:`2617`).

"""

##########################################################################
Expand Down Expand Up @@ -1089,29 +1100,27 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option')
if func == Enum_QSFunction.Quick_Start_Request:
schema_req = cast('Schema_QuickStartRequestOption', schema)

rate = schema_req.flags['rate']
opt = Data_QuickStartRequestOption(
type=schema.type,
action=Enum_OptionAction.get(schema.type >> 6),
change=bool(schema.type & 0b00100000),
length=self._ipv6_opts_option_length(schema_req.len),
func=func,
rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0,
rate=self._read_qs_rate(schema_req.flags['rate']),
ttl=datetime.timedelta(seconds=schema_req.ttl),
nonce=schema_req.nonce['nonce'],
reserved=schema_req.nonce['reserved'],
) # type: Data_QuickStartOption
elif func == Enum_QSFunction.Report_of_Approved_Rate:
schema_rep = cast('Schema_QuickStartReportOption', schema)

rate = schema_rep.flags['rate']
opt = Data_QuickStartReportOption(
type=schema.type,
action=Enum_OptionAction.get(schema.type >> 6),
change=bool(schema.type & 0b00100000),
length=self._ipv6_opts_option_length(schema_rep.len),
func=func,
rate=40000 * (2 ** rate) // 1000 if rate > 0 else 0,
rate=self._read_qs_rate(schema_rep.flags['rate']),
nonce=schema_rep.nonce['nonce'],
reserved=schema_rep.nonce['reserved'],
unused=schema_rep.reserved,
Expand Down Expand Up @@ -1897,20 +1906,24 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption
func_default: default value for QS function type
func_namespace: namespace for QS function type
func_reversed: reversed flag for QS function type
rate: rate (in kbps)
rate: rate (in kbps), rounded down to the nearest rate the
four-bit rate code carries (:rfc:`4782#section-3.1`)
ttl: time to live (in seconds)
nonce: nonce value
reserved: reserved bits (``R``) after the nonce
unused: the "Not Used" octet of a report of approved rate
**kwargs: arbitrary keyword arguments

Returns:
Constructed option schema; an unassigned option schema if ``opt``
is a Quick-Start option parsed as unassigned, see :meth:`_read_opt_qs`.
Constructed option schema. An unassigned option schema if ``opt``
is a Quick-Start option parsed as unassigned, see :meth:`_read_opt_qs`,
or if ``func`` is neither of the assigned functions, in which case
the option data takes the layout of a Quick-Start Request.

Raises:
ProtocolError: If the QS function is neither a Quick-Start Request
nor a Report of Approved Rate.
ProtocolError: If ``func`` does not fit in four bits, the rate code
cannot carry ``rate``, or ``ttl``, ``nonce`` or ``reserved`` does
not fit its field.

"""
if isinstance(opt, Data_UnassignedOption):
Expand All @@ -1925,19 +1938,20 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption
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
rate_val = self._make_qs_rate(rate)
ttl_value = ttl if isinstance(ttl, int) else math.floor(ttl.total_seconds())
nonce = self._make_uint(nonce, bits=30, name='Quick-Start nonce')
reserved = self._make_uint(reserved, bits=2, name='Quick-Start reserved')

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=code,
len=6,
flags={
'func': func_enum,
'rate': rate_val,
},
ttl=ttl_value,
ttl=self._make_uint(ttl_value, bits=8, name='Quick-Start TTL'),
nonce={
'nonce': nonce,
'reserved': reserved,
Expand All @@ -1957,7 +1971,11 @@ def _make_opt_qs(self, code: 'Enum_Option', opt: 'Optional[Data_QuickStartOption
'reserved': reserved,
},
)
raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid QS function: {func_enum}')
if not 0 <= func_enum < 16:
raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid QS function: {func_enum}')
ttl_value = self._make_uint(ttl_value, bits=8, name='Quick-Start TTL')
data = bytes([(func_enum << 4) | rate_val, ttl_value]) + ((nonce << 2) | reserved).to_bytes(4, 'big')
return self._make_opt_none(code, data=data)

def _make_opt_rpl(self, code: 'Enum_Option', opt: 'Optional[Data_RPLOption]' = None, *,
down: 'bool' = False,
Expand Down
49 changes: 49 additions & 0 deletions pcapkit/protocols/protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -2654,6 +2654,55 @@ def _check_term_threshold(self) -> bool:

return layer_match or protocol_match

def _read_qs_rate(self, code: 'int') -> 'int':
"""Read a Quick-Start rate code as a rate in kbps.

:rfc:`4782#section-3.1` encodes the Rate Request, and the Rate Report
formatted exactly as it, as a four-bit ``N`` that stands for
``K * 2 ** N`` bps with ``K`` set to 40,000, and ``N = 0`` for a rate of
zero. The IPv6 option keeps the IPv4 format (:rfc:`4782#section-3.2`),
and the TCP Quick-Start Response formats its Rate Request the same way
(:rfc:`4782#section-4.2`).

Args:
code: the four-bit rate field, ``N``

Returns:
The rate in kbps: ``0`` for ``N = 0``, else ``40 * 2 ** N``.

.. versionadded:: 1.6.0

"""
return 40 * 2 ** code if code else 0

def _make_qs_rate(self, rate: 'Any') -> 'int':
"""Make the Quick-Start rate code for a rate in kbps.

The inverse of :meth:`_read_qs_rate`. A rate between two codes is
rounded down to the lower one, so the option never carries more than
the rate given.

Args:
rate: the rate, in kbps

Returns:
The four-bit rate field, ``N``.

Raises:
ProtocolError: If ``rate`` is not an :class:`int`, is a
:obj:`bool`, or is one the field cannot carry: negative, above
zero but below the 80 kbps of ``N = 1``, or at or above the
``40 * 2 ** 16`` kbps that needs a fifth bit. The message names
:attr:`alias`.

.. versionadded:: 1.6.0

"""
if (not isinstance(rate, int) or isinstance(rate, bool)
or (rate and not 80 <= rate < 40 * 2 ** 16)):
raise ProtocolError(f'{self.alias}: invalid Quick-Start rate: {rate!r}')
return (rate // 40).bit_length() - 1 if rate else 0


class Protocol(ProtocolBase, Generic[_PT, _ST]):
"""User-facing abstract base class for third-party protocols.
Expand Down
Loading
Loading