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
2 changes: 1 addition & 1 deletion pcapkit/protocols/internet/hopopt.py
Original file line number Diff line number Diff line change
Expand Up @@ -934,7 +934,7 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option')
if schema.len != 6:
raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format')

func = schema.func
func = Enum_QSFunction.get(schema.flags['func'])
if func == Enum_QSFunction.Quick_Start_Request:
schema_req = cast('Schema_QuickStartRequestOption', schema)

Expand Down
2 changes: 1 addition & 1 deletion pcapkit/protocols/internet/ipv4.py
Original file line number Diff line number Diff line change
Expand Up @@ -1178,7 +1178,7 @@ def _read_opt_qs(self, schema: 'Schema_QSOption', *, options: 'Option') -> 'Data
if schema.length != 8:
raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format')

func = schema.func
func = Enum_QSFunction.get(schema.flags['func'])
if func == Enum_QSFunction.Quick_Start_Request:
schema_req = cast('Schema_QuickStartRequestOption', schema)

Expand Down
2 changes: 1 addition & 1 deletion pcapkit/protocols/internet/ipv6_opts.py
Original file line number Diff line number Diff line change
Expand Up @@ -923,7 +923,7 @@ def _read_opt_qs(self, schema: 'Schema_QuickStartOption', *, options: 'Option')
if schema.len != 6:
raise ProtocolError(f'{self.alias}: [OptNo {schema.type}] invalid format')

func = schema.func
func = Enum_QSFunction.get(schema.flags['func'])
if func == Enum_QSFunction.Quick_Start_Request:
schema_req = cast('Schema_QuickStartRequestOption', schema)

Expand Down
6 changes: 5 additions & 1 deletion pcapkit/protocols/schema/internet/hopopt.py
Original file line number Diff line number Diff line change
Expand Up @@ -252,7 +252,11 @@ def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field':
schema = QuickStartOption.registry[func]
if schema is None:
raise FieldValueError(f'HOPOPT: invalid QS function: {func}')
return SchemaField(length=5, schema=schema)

# NOTE: The suboption schemas carry the option's own ``type`` and ``len``
# octets, so this is the width of the whole option: the two header octets
# plus the six data octets :rfc:`4782#section-3.2` gives both functions.
return SchemaField(length=8, schema=schema)


def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int':
Expand Down
6 changes: 5 additions & 1 deletion pcapkit/protocols/schema/internet/ipv6_opts.py
Original file line number Diff line number Diff line change
Expand Up @@ -252,7 +252,11 @@ def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field':
schema = QuickStartOption.registry[func]
if schema is None:
raise FieldValueError(f'IPv6-Opts: invalid QS function: {func}')
return SchemaField(length=5, schema=schema)

# NOTE: The suboption schemas carry the option's own ``type`` and ``len``
# octets, so this is the width of the whole option: the two header octets
# plus the six data octets :rfc:`4782#section-3.2` gives both functions.
return SchemaField(length=8, schema=schema)


def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int':
Expand Down
3 changes: 1 addition & 2 deletions tests/protocols/internet/test_ipv4_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -1859,11 +1859,10 @@ def assert_bad(reader, schema) -> None:
qs_unknown = QuickStartRequestOption(
type=OptionNumber.QS,
length=8,
flags={'func': QSFunction.Quick_Start_Request, 'rate': 1},
flags={'func': QSFunction.get(1), 'rate': 1},
ttl=7,
nonce={'nonce': 3},
)
object.__setattr__(qs_unknown, 'func', QSFunction.get(1))
assert_bad(proto._read_opt_qs, qs_unknown)

def test_ipv4_schema_helpers_and_post_process_branches(self) -> None:
Expand Down
3 changes: 1 addition & 2 deletions tests/protocols/internet/test_ipv6_extension_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -1330,11 +1330,10 @@ def assert_bad(reader, option_schema) -> None:
qs_unknown = schema.QuickStartRequestOption(
type=Option.Quick_Start,
len=6,
flags={'func': QSFunction.Quick_Start_Request, 'rate': 1},
flags={'func': QSFunction.get(1), 'rate': 1},
ttl=7,
nonce={'nonce': 3},
)
object.__setattr__(qs_unknown, 'func', QSFunction.get(1))
assert_bad(proto._read_opt_qs, qs_unknown)
assert_bad(proto._read_opt_rpl, schema.RPLOption(
type=Option.RPL_Option_0x63, len=3,
Expand Down
105 changes: 105 additions & 0 deletions tests/protocols/internet/test_quick_start_func_unit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,105 @@
# -*- coding: utf-8 -*-
"""Quick-Start options in IPv4, HOPOPT and IPv6-Opts build, parse and round-trip.

GitHub issue #1091. ``_read_opt_qs`` read ``schema.func``, which the schema
declares only under ``TYPE_CHECKING`` and which only ``_QSOption.post_process``
sets on the parse path. A schema built by ``_make_opt_qs`` never passes through
that hook, so constructing any Quick-Start option raised ``AttributeError:
'QuickStartRequestOption' object has no attribute 'func'``. The readers now
take the function from ``flags['func']``, the bit-field they already read
``rate`` from.

HOPOPT and IPv6-Opts also sized the nested suboption with
``SchemaField(length=5)``, three octets short of the eight-octet option, so a
well-formed option decoded its nonce wrongly and left octets behind;
:meth:`QuickStartWireTests.test_wire_option_decodes_the_full_nonce` covers that.

"""

from __future__ import annotations

import datetime
import importlib.util
import unittest
import warnings

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)

#: A 30-bit nonce whose wire form, shifted left two bits, is ``deadbee0``.
NONCE = 0xDEADBEE0 >> 2


def _families() -> 'list[tuple]':
from pcapkit.const.ipv4.option_number import OptionNumber as Enum_IPv4Option
from pcapkit.const.ipv6.option import Option as Enum_IPv6Option
from pcapkit.const.reg.transtype import TransType
from pcapkit.protocols.internet.hopopt import HOPOPT
from pcapkit.protocols.internet.ipv4 import IPv4
from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts

return [
('IPv4', IPv4, dict(src='192.0.2.1', dst='192.0.2.2', protocol=TransType.UDP),
Enum_IPv4Option.QS),
('HOPOPT', HOPOPT, dict(next=TransType.UDP), Enum_IPv6Option.Quick_Start),
('IPv6-Opts', IPv6_Opts, dict(next=TransType.UDP), Enum_IPv6Option.Quick_Start),
]


@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
class QuickStartRoundTripTests(unittest.TestCase):
def setUp(self) -> None:
reimport_once_per_class(self)

def test_quick_start_option_round_trips(self) -> None:
from pcapkit.const.ipv6.qs_function import QSFunction as Enum_QSFunction

for name, cls, base, code in _families():
for func in (Enum_QSFunction.Quick_Start_Request,
Enum_QSFunction.Report_of_Approved_Rate):
with self.subTest(family=name, func=func.name):
with warnings.catch_warnings():
warnings.simplefilter('error')
built = cls(options=[(code, dict(func=func, rate=80, ttl=5, nonce=NONCE))],
**base)
again = cls(built.data)
self.assertEqual(again.data, built.data)

option = list(again.info.options.values())[0]
self.assertEqual(option.func, func)
self.assertEqual(option.rate, 80)
self.assertEqual(option.nonce, NONCE)
if func == Enum_QSFunction.Quick_Start_Request:
self.assertEqual(option.ttl, datetime.timedelta(seconds=5))


@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
class QuickStartWireTests(unittest.TestCase):
def setUp(self) -> None:
reimport_once_per_class(self)

def test_wire_option_decodes_the_full_nonce(self) -> None:
from pcapkit.const.ipv6.qs_function import QSFunction as Enum_QSFunction
from pcapkit.protocols.internet.hopopt import HOPOPT
from pcapkit.protocols.internet.ipv6_opts import IPv6_Opts

# next=UDP, len=1 (16 octets); Quick-Start Request type 0x26 len 6,
# func 0 rate 1, ttl 5, nonce; then PadN of four octets.
wire = bytes.fromhex('1101' '26060105deadbee0' '010400000000')
for cls in (HOPOPT, IPv6_Opts):
with self.subTest(family=cls.__name__):
with warnings.catch_warnings():
warnings.simplefilter('error')
parsed = cls(wire)

options = list(parsed.info.options.values())
self.assertEqual(options[0].func, Enum_QSFunction.Quick_Start_Request)
self.assertEqual(options[0].nonce, NONCE)
self.assertEqual(options[0].ttl, datetime.timedelta(seconds=5))
self.assertEqual(len(options), 2)


if __name__ == '__main__':
unittest.main()
56 changes: 10 additions & 46 deletions tests/protocols/test_option_roundtrip_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -246,52 +246,16 @@ class Gap(NamedTuple):

# -- Quick-Start, in all three protocols that carry it --------------------

# ``_make_opt_qs`` returns a bare nested schema, but ``func`` is set only by
# ``_QSOption.post_process``, which runs on the parse path. Since
# ``__post_init__`` packs and then re-reads, construction fails.
#
# There was a second, independent defect in the same option that these cases
# never reach, and it was the more serious of the two:
# ``quick_start_data_selector`` handed the nested schema a hardcoded
# ``SchemaField(length=5)`` where a Quick-Start Request needs eight octets
# (type 1 + length 1 + flags 1 + ttl 1 + nonce 4). Measured on a
# hand-built, well-formed 8-octet IPv4 Quick-Start option
# ``1908002adeadbee0``: it parsed "successfully" with
# ``SchemaWarning: packet length < 0: -3`` and decoded ``nonce`` as **55**
# instead of 933982136 -- silent corruption rather than a failure -- and the
# three unconsumed octets were then read as a further, fabricated option,
# which made the enclosing IPv4 datagram fail with ``ProtocolError: IPv4:
# invalid format``.
#
# #552 fixed IPv4's copy: the length now comes from
# ``quick_start_option_length(schema)``, i.e. from the suboption the selector
# resolved, and ``QuickStartReportOption`` gained the ``Not Used`` octet
# :rfc:`4782#section-3.1` gives it and it was missing -- it packed seven
# octets against the ``length=8`` that ``_make_opt_qs`` writes and
# ``_read_opt_qs`` demands, so a spec-correct Report of Approved Rate read off
# the wire decoded its nonce one octet early. ``ipv4-option/QS`` is still
# recorded below because the ``func`` defect is untouched and fails first.
#
# The identical ``SchemaField(length=5)`` is still at
# ``schema/internet/hopopt.py:255`` and ``schema/internet/ipv6_opts.py:255``,
# with the same measured nonce of 55; #552 was scoped to IPv4's copy. Note a
# fix there is not a copy of this one: :rfc:`4782#section-3.2` sets the IPv6
# option's ``length`` field to 6 rather than 8, since it excludes the common
# type and length octets the extension header already carries. Fixing
# the ``func`` defect alone will not make any of these three cases pass.
'ipv4-option/QS': Gap(
'CONSTRUCT', "no attribute 'func'",
'pcapkit/protocols/internet/ipv4.py:1178 -- func is set only by '
'post_process'),
'hopopt-option/Quick_Start': Gap(
'CONSTRUCT', "no attribute 'func'",
'pcapkit/protocols/internet/hopopt.py:918; and separately '
'pcapkit/protocols/schema/internet/hopopt.py:255 -- SchemaField(length=5)'),
'ipv6-opts-option/Quick_Start': Gap(
'CONSTRUCT', "no attribute 'func'",
'pcapkit/protocols/internet/ipv6_opts.py:921; and separately '
'pcapkit/protocols/schema/internet/ipv6_opts.py:255 -- '
'SchemaField(length=5)'),
# ``ipv4-option/QS``, ``hopopt-option/Quick_Start`` and
# ``ipv6-opts-option/Quick_Start`` used to be recorded here. ``_read_opt_qs``
# read ``schema.func``, which is declared only under ``TYPE_CHECKING`` and set
# only by ``_QSOption.post_process`` on the parse path, so the schema
# ``_make_opt_qs`` builds raised ``AttributeError`` when ``__post_init__``
# re-read it. The readers now take the function from ``flags['func']``, as
# they already took ``rate`` (#1091). HOPOPT and IPv6-Opts also sized the
# nested schema with ``SchemaField(length=5)`` against an eight-octet option,
# which decoded the nonce wrongly; #552 had fixed only IPv4's copy, and #1091
# fixes the other two. All three now round-trip.

# -- The non-progress loop, now fully fixed -------------------------------

Expand Down
Loading