diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 1520944bd1..fd04e0d45b 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -12,6 +12,7 @@ """ import abc +import ast import collections import contextlib import difflib @@ -30,6 +31,7 @@ import aenum from pcapkit.corekit.context import ContextRegistry +from pcapkit.corekit.fields.misc import ConditionalField, SchemaField, SwitchField from pcapkit.corekit.module import ModuleDescriptor from pcapkit.corekit.protochain import ProtoChain from pcapkit.corekit.sentinels import ABSENT, AbsentType # pylint: disable=unused-import @@ -43,8 +45,9 @@ from pcapkit.utilities.chardet import detect from pcapkit.utilities.compat import cached_property from pcapkit.utilities.decorators import beholder, seekset -from pcapkit.utilities.exceptions import (ProtocolNotFound, ProtocolNotImplemented, RegistryError, - StructError, UnsupportedCall) +from pcapkit.utilities.exceptions import (BaseError, ProtocolError, ProtocolNotFound, + ProtocolNotImplemented, RegistryError, StructError, + UnsupportedCall) from pcapkit.utilities.warnings import RegistryWarning, UnknownFieldWarning, warn if TYPE_CHECKING: @@ -67,6 +70,128 @@ # readable characters' order list readable = [ord(char) for char in filter(lambda char: not char.isspace(), string.printable)] + +@functools.lru_cache(maxsize=None) +def _make_index_calls(func: 'Callable[..., Any]') -> 'tuple[tuple[int, int, str], ...]': + """Locate the ``_make_index`` calls in the source of ``func``. + + Args: + func: method calling ``_make_index`` + + Returns: + One ``(first_line, last_line, argument)`` triple per call whose first + argument is a plain name, with absolute line numbers. Empty if the + source of ``func`` is unavailable. + + """ + try: + lines, start = inspect.getsourcelines(func) + tree = ast.parse(textwrap.dedent(''.join(lines))) + except (OSError, TypeError, SyntaxError): + return () + return tuple( + (node.lineno + start - 1, getattr(node, 'end_lineno', node.lineno) + start - 1, node.args[0].id) + for node in ast.walk(tree) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + and node.func.attr == '_make_index' and node.args and isinstance(node.args[0], ast.Name) + ) + + +def _passes_raw(value: 'Any', nested: 'bool' = False) -> 'bool': + """Tell if ``make()`` keywords may carry a raw :obj:`bytes` item. + + Args: + value: keyword arguments of ``make()``, or a value within them + nested: if ``value`` sits inside a container + + Returns: + :data:`True` if :obj:`bytes` sits anywhere inside a :obj:`list`, + :obj:`tuple` or :obj:`dict` among ``value``, or a + :class:`~pcapkit.protocols.schema.schema.Schema` does anywhere. A + top-level :obj:`bytes` keyword (e.g. a payload) does not count. + + """ + if isinstance(value, bytes): + return nested + if isinstance(value, Schema): # a ready-made item, which ``make()`` may pack into raw octets + return True + if isinstance(value, dict): + return any(_passes_raw(item, nested) for key, item in value.items() if key != '__packet__') + if isinstance(value, (list, tuple)): + return any(_passes_raw(item, True) for item in value) + return False + + +@functools.lru_cache(maxsize=None) +def _schema_slots(schema: 'Type[Schema]') -> 'tuple[str, ...]': + """Names of the fields of ``schema`` that hold a schema. + + Args: + schema: schema class + + Returns: + Names of the :class:`~pcapkit.corekit.fields.misc.SchemaField` and + :class:`~pcapkit.corekit.fields.misc.SwitchField` fields, including + those wrapped in a :class:`~pcapkit.corekit.fields.misc.ConditionalField`. + + """ + slots = [] + for name, field in schema.__fields__.items(): + while isinstance(field, ConditionalField): + field = field.field + if isinstance(field, (SchemaField, SwitchField)): + slots.append(name) + return tuple(slots) + + +def _holds_raw(value: 'Any') -> 'bool': + """Tell if a made schema may hold raw :obj:`bytes` in place of a schema. + + Args: + value: made schema, or a value within it + + Returns: + :data:`True` if ``value`` holds :obj:`bytes` as a list item, or as + the value of a field that holds a schema + (:class:`~pcapkit.corekit.fields.misc.SchemaField` or + :class:`~pcapkit.corekit.fields.misc.SwitchField`). + + """ + if isinstance(value, (list, tuple)): + return any(isinstance(item, bytes) or _holds_raw(item) for item in value) + if isinstance(value, Schema): + for name, field in value.__fields__.items(): + item = value.get(name) + while isinstance(field, ConditionalField): + field = field.field + if isinstance(item, bytes) and isinstance(field, (SchemaField, SwitchField)): + return True + if _holds_raw(item): + return True + return False + + +def _raw_where_parsed(made: 'Any', parsed: 'Any') -> 'bool': + """Tell if ``made`` holds raw :obj:`bytes` where ``parsed`` holds a schema. + + Args: + made: made schema, or a value within it + parsed: the same position in the schema parsed from its packed octets + + Returns: + :data:`True` if, at some position both share, ``made`` holds + :obj:`bytes` and ``parsed`` a :class:`~pcapkit.protocols.schema.schema.Schema`. + + """ + if isinstance(made, bytes): + return isinstance(parsed, Schema) + if isinstance(made, (list, tuple)) and isinstance(parsed, (list, tuple)): + return any(_raw_where_parsed(item, other) for item, other in zip(made, parsed)) + if isinstance(made, Schema) and isinstance(parsed, Schema): + return any(_raw_where_parsed(made.get(name), parsed.get(name)) + for name in made.__fields__ if name in parsed.__fields__) + return False + #: Keywords that configure the construction rather than naming a field, and are #: therefore consumed by :meth:`ProtocolBase.__init__ #: ` or by the schema layer @@ -572,6 +697,9 @@ def pack(self, **kwargs: 'Any') -> 'bytes': """ self.__header__ = self.make(**kwargs) + #: bool: If the made header may hold raw items (see :meth:`unpack`). + self._make_raw = _passes_raw(kwargs) or any( + isinstance(self.__header__.get(name), bytes) for name in _schema_slots(type(self.__header__))) packet = kwargs.get('__packet__', {}) # packet data return self.__header__.pack(packet) @@ -590,10 +718,28 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT': global packet data to underlying methods. This is useful when the packet data is not available in the current instance. + When :meth:`pack` or :meth:`from_schema` has already set + :attr:`self.__header__ `, that schema is + kept, unless it still holds raw :obj:`bytes` where parsing its + packed octets yields a schema -- e.g. a raw option given to + ``make()`` -- which :meth:`read` cannot consume. The header is + then replaced by the parsed one (:issue:`1312`). + + Raises: + ProtocolError: If the made header holds raw items and its packed + octets do not parse. + """ + packet = kwargs.get('__packet__', {}) # packet data if cast('Optional[_ST]', self.__header__) is None: - packet = kwargs.get('__packet__', {}) # packet data self.__header__ = cast('_ST', self.__schema__.unpack(self._file, length, packet)) # type: ignore[call-arg,misc] + elif self._make_raw and _holds_raw(self.__header__): + try: + parsed = cast('_ST', type(self.__header__).unpack(io.BytesIO(self._data), length, packet)) # type: ignore[call-arg,misc] + except (BaseError, struct.error) as error: # pylint: disable=no-member + raise ProtocolError(f'{type(self).__name__}: malformed raw item given to make()') from error + if _raw_where_parsed(self.__header__, parsed): + self.__header__ = parsed return self.read(length, **kwargs) @staticmethod @@ -857,6 +1003,9 @@ def __new__(cls, *args: 'Any', **kwargs: 'Any') -> 'Self': # pylint: disable=un # reference between instances. self.__cached__ = {} self.__header__ = None # type: ignore[assignment] + #: bool: If a header set before :meth:`unpack` may hold raw items; + #: :meth:`pack` narrows it down for a made one. + self._make_raw = True return self @@ -1603,12 +1752,8 @@ def _make_index(cls, name: 'str | int | StdlibEnum | AenumEnum', default: 'Optio If ``name`` is a :obj:`str` and ``namespace`` is :data:`None`, the namespace is inferred from the calling ``make`` method: it is the enumeration class of the default value of the argument - that ``name`` was passed as. If several arguments were given the - same name, every one of their enumerations must have the name - and map it to the same value; otherwise - :exc:`~pcapkit.utilities.exceptions.ProtocolNotImplemented` is - raised, asking for an explicit ``*_namespace`` (see - :meth:`_make_index_namespace`). + that ``name`` was passed as, found from the call in the + caller's source (see :meth:`_make_index_namespace`). """ if isinstance(name, (enum.Enum, aenum.Enum)): @@ -1659,6 +1804,15 @@ def _make_index_namespace(cls, name: 'str', member. The candidate namespace of a match is the class of that default value. + Only the argument the call passes as ``name`` is considered: the + caller's source is parsed (by ``_make_index_calls``) for the + :meth:`_make_index` call spanning the current line, and its first + argument names the method argument, so two arguments holding the + same member name each resolve against their own enumeration. Every + argument is considered when that call cannot be located (e.g. the + source is unavailable) or its first argument is not an argument of + the method (e.g. ``value = arg; self._make_index(value)``). + Args: name: member name given to :meth:`_make_index` frame: frame of :meth:`_make_index` @@ -1671,9 +1825,9 @@ def _make_index_namespace(cls, name: 'str', candidate has ``name`` as a member. Raises: - ProtocolNotImplemented: If several arguments match and their - candidates disagree on ``name``: some lack it, or they map - it to different values. + ProtocolNotImplemented: If every argument is considered, several + arguments match and their candidates disagree on ``name``: + some lack it, or they map it to different values. """ caller = frame.f_back if frame is not None else None @@ -1696,8 +1850,13 @@ def _make_index_namespace(cls, name: 'str', return None params = inspect.signature(func).parameters + lineno = caller.f_lineno + located = {arg for first, last, arg in _make_index_calls(func) if first <= lineno <= last} found = {} # type: dict[str, Type[StdlibEnum] | Type[AenumEnum]] for arg, param in params.items(): + # once the call is located, only its own argument is a candidate + if len(located) == 1 and located <= params.keys() and arg not in located: + continue value = values.get(arg) if f'{arg}_namespace' not in params or not isinstance(value, str) or value != name: continue diff --git a/tests/protocols/test_make_index_str_unit.py b/tests/protocols/test_make_index_str_unit.py index cd4beb70d0..ea8ccae827 100644 --- a/tests/protocols/test_make_index_str_unit.py +++ b/tests/protocols/test_make_index_str_unit.py @@ -108,7 +108,8 @@ def test_shared_name_with_one_meaning_resolves(self) -> None: self.assertEqual(ipv4(tos_thr='HIGH', tos_rel='HIGH').data, expected) self.assertNotEqual(ipv4().data, expected) - def test_shared_name_with_two_meanings_raises(self) -> None: + def test_shared_name_with_two_meanings_resolves_per_argument(self) -> None: + # the per-argument resolution of #1308; see test_protocol_make_unit enum = importlib.import_module('enum') exc = _attr('pcapkit.utilities.exceptions.ProtocolNotImplemented') protocol = _attr('pcapkit.protocols.protocol.ProtocolBase') @@ -121,16 +122,13 @@ def make_pair(self, a: 'Any' = left.Shared, a_namespace: 'Any' = None, b: 'Any' = right.Shared, b_namespace: 'Any' = None) -> 'int': return self._make_index(a, namespace=a_namespace) - with self.assertRaisesRegex(exc, r"ambiguous member name 'Shared'.*" - r"pass a_namespace= or b_namespace= explicitly"): - Pair.make_pair(Pair, a='Shared', b='Shared') - # an explicit namespace on the other argument removes the ambiguity + self.assertEqual(Pair.make_pair(Pair, a='Shared', b='Shared'), 1) self.assertEqual(Pair.make_pair(Pair, a='Shared', b='Shared', b_namespace=right), 1) - # a real make(): ``LOW`` is a ToSDelay member but not a ToSThroughput one + # a real make(): ``LOW`` is a ToSDelay member but not a ToSThroughput one, + # so ``tos_thr`` alone fails to resolve it ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4') - with self.assertRaisesRegex(exc, r"ambiguous member name 'LOW'.*" - r"pass tos_del_namespace= or tos_thr_namespace= explicitly"): + with self.assertRaisesRegex(exc, r"protocol 'LOW' not implemented"): ipv4(tos_del='LOW', tos_thr='LOW') def test_without_inferable_enum_still_raises(self) -> None: diff --git a/tests/protocols/test_protocol_make_unit.py b/tests/protocols/test_protocol_make_unit.py new file mode 100644 index 0000000000..d157318620 --- /dev/null +++ b/tests/protocols/test_protocol_make_unit.py @@ -0,0 +1,236 @@ +# -*- coding: utf-8 -*- +"""Construction through ``make()`` in :class:`~pcapkit.protocols.protocol.Protocol`. + +GitHub issue #1312: :meth:`~pcapkit.protocols.protocol.ProtocolBase.unpack` +reused the schema :meth:`~pcapkit.protocols.protocol.ProtocolBase.pack` had +made instead of parsing the packed octets, so ``read`` met the raw +:obj:`bytes` (or :class:`~pcapkit.protocols.schema.schema.Schema`) items +the option makers accept, and construction crashed with +:exc:`AttributeError`. A bare ``MH()`` crashed the same way on its default +``data=b'\\x00\\x00'``. A made schema holding raw octets where parsing its +packed form yields a schema is now replaced by the parsed one. + +GitHub issue #1308: a :obj:`str` member name given to two enum arguments +of one ``make()`` raised ``ambiguous member name`` when only one of the +enumerations had it. Each ``_make_index`` call now resolves against the +argument it was passed. + +Every case builds its own packet in memory and reads no capture. Classes are +imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, so they belong to the live +:mod:`pcapkit` import. + +""" + +import importlib +import unittest +import unittest.mock +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + + +def _attr(path: 'str') -> 'Any': + module, _, name = path.rpartition('.') + return getattr(importlib.import_module(module), name) + + +#: One raw item per protocol whose ``make()`` accepts raw octets in a list: +#: (protocol class, construction keywords, packed octets). +RAW_ITEMS = { + 'HIP parameters': ('pcapkit.protocols.internet.hip.HIP', + {'extension': True, 'parameters': [b'\x00\x41\x00\x0c' + bytes(12)]}, + '1106206100000000' + '00' * 32 + '0041000c' + '00' * 12), + 'IPv4 options': ('pcapkit.protocols.internet.ipv4.IPv4', + {'options': [b'\x94\x04\x00\x00'], 'protocol': 253}, + '460000180000000000fd00007f0000010000000094040000'), + 'TCP options': ('pcapkit.protocols.transport.tcp.TCP', + {'options': [b'\x02\x04\x05\xb4']}, + '0000000000000000000000006000ffff00000000020405b4'), + 'HOPOPT options': ('pcapkit.protocols.internet.hopopt.HOPOPT', + {'extension': True, 'options': [b'\x05\x02\x00\x00']}, + '1100050200000000'), + 'IPv6_Opts options': ('pcapkit.protocols.internet.ipv6_opts.IPv6_Opts', + {'extension': True, 'options': [b'\x05\x02\x00\x00']}, + '1100050200000000'), + 'MH options': ('pcapkit.protocols.internet.mh.MH', + {'data': {'options': [b'\x01\x02\x00\x00']}}, + '11010000000000000102000001020000'), + 'SCTP chunks': ('pcapkit.protocols.transport.sctp.SCTP', + {'chunks': [b'\x0e\x00\x00\x04']}, + '0000000000000000617cebb70e000004'), + 'IPv6_Route data': ('pcapkit.protocols.internet.ipv6_route.IPv6_Route', + {'extension': True, 'data': bytes(4)}, + '1100000000000000'), +} + + +class TestMakeRawItems(unittest.TestCase): + """Raw and :class:`Schema` items given to ``make()`` build a packet (#1312).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_raw_item_builds_and_parses(self) -> None: + for case, (klass, kwargs, packed) in RAW_ITEMS.items(): + with self.subTest(case=case): + cls = _attr(klass) + proto = cls(**kwargs) + self.assertEqual(proto.data.hex(), packed) + # the made packet reads as the same octets parsed afresh + extra = {'extension': True} if kwargs.get('extension') else {} + self.assertEqual(proto.info, cls(proto.data, **extra).info) + + def test_schema_item_builds_as_raw_item(self) -> None: + for case in ('HIP parameters', 'IPv4 options', 'TCP options'): + with self.subTest(case=case): + klass, kwargs, packed = RAW_ITEMS[case] + cls = _attr(klass) + key = next(key for key in ('parameters', 'options') if key in kwargs) + extra = {'extension': True} if kwargs.get('extension') else {} + header = cls(bytes.fromhex(packed), **extra).__header__ + item = getattr(header, 'param' if key == 'parameters' else 'options')[0] + self.assertNotIsInstance(item, bytes) + proto = cls(**{**kwargs, key: [item]}) + self.assertEqual(proto.data.hex(), packed) + + def test_bare_mh_constructs(self) -> None: + mh = _attr('pcapkit.protocols.internet.mh.MH') + proto = mh() + self.assertEqual(proto.data.hex(), '1100000000000000') + self.assertEqual(proto.info, mh(proto.data).info) + + def test_made_schema_without_raw_items_is_kept(self) -> None: + module = importlib.import_module('pcapkit.protocols.protocol') + ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4') + option = _attr('pcapkit.const.ipv4.option_number.OptionNumber') + + # an option given as a tuple is made into a schema, so nothing is re-parsed + made = ipv4.make(ipv4.__new__(ipv4), options=[(option.RTRALT, {'alert': 0})], protocol=253) + self.assertFalse(module._holds_raw(made)) + raw = ipv4.make(ipv4.__new__(ipv4), options=[b'\x94\x04\x00\x00'], protocol=253) + self.assertTrue(module._holds_raw(raw)) + + parsed = ipv4(raw.pack()).__header__ + self.assertTrue(module._raw_where_parsed(raw, parsed)) + self.assertFalse(module._raw_where_parsed(parsed, parsed)) + # raw octets matched by raw octets are not a stand-in for a schema + self.assertFalse(module._raw_where_parsed([b'\x01'], [b'\x01'])) + + def test_unparsable_raw_item_raises_protocol_error(self) -> None: + exc = _attr('pcapkit.utilities.exceptions.ProtocolError') + ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4') + # a lone length-less option makes the packed octets unparsable + with self.assertRaisesRegex(exc, r'malformed raw item given to make\(\)') as caught: + ipv4(options=[b'\x02', b'\x94\x04\x00\x00'], protocol=253) + self.assertIsInstance(caught.exception.__cause__, exc) + + def test_random_raw_items_raise_library_errors_only(self) -> None: + random = importlib.import_module('random') + base = _attr('pcapkit.utilities.exceptions.BaseError') + rng = random.Random(1352) + + def items() -> 'list[bytes]': + return [bytes(rng.randrange(256) for _ in range(rng.randrange(1, 13))) + for _ in range(rng.randrange(1, 3))] + + for case, (klass, kwargs, _) in RAW_ITEMS.items(): + key = next(iter(kw for kw in ('parameters', 'options', 'chunks', 'data') if kw in kwargs)) + cls = _attr(klass) + with self.subTest(case=case): + for _ in range(200): + value = items() # type: Any + if case == 'MH options': + value = {'options': value} + elif case == 'IPv6_Route data': + value = b''.join(value) + try: + with importlib.import_module('warnings').catch_warnings(): + importlib.import_module('warnings').simplefilter('ignore') + cls(**{**kwargs, key: value}) + except base: + pass + + def test_malformed_raw_item_raises_protocol_error(self) -> None: + # ESP_INFO needs a 12-octet body; an 8-octet one is a parse error, + # reported as such rather than as an AttributeError on raw bytes + hip = _attr('pcapkit.protocols.internet.hip.HIP') + exc = _attr('pcapkit.utilities.exceptions.ProtocolError') + with self.assertRaisesRegex(exc, r'\[ParamNo 65\] invalid format'): + hip(extension=True, parameters=[b'\x00\x41\x00\x08' + bytes(8)]) + + +class TestMakeIndexPerArgument(unittest.TestCase): + """A member name resolves against its own ``make()`` argument (#1308).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _pair(self) -> 'tuple[Any, Any, Any]': + enum = importlib.import_module('enum') + protocol = _attr('pcapkit.protocols.protocol.ProtocolBase') + + left = enum.IntEnum('Left', {'Shared': 1}) + right = enum.IntEnum('Right', {'Shared': 2, 'Only': 3}) + + class Pair(protocol): # type: ignore[misc,valid-type] + def make_pair(self, a: 'Any' = left.Shared, a_namespace: 'Any' = None, + b: 'Any' = right.Shared, b_namespace: 'Any' = None) -> 'tuple[int, int]': + return (self._make_index(a, namespace=a_namespace), + self._make_index(b, 0, namespace=b_namespace)) + + return Pair, left, right + + def test_shared_name_resolves_per_argument(self) -> None: + pair, _, _ = self._pair() + self.assertEqual(pair.make_pair(pair, a='Shared', b='Shared'), (1, 2)) + # a name only one argument holds resolves as it did before + self.assertEqual(pair.make_pair(pair, a='Shared', b='Only'), (1, 3)) + + def test_ipv4_shared_name_falls_back_per_argument(self) -> None: + ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4') + tos_del = _attr('pcapkit.const.ipv4.tos_del.ToSDelay') + + expected = ipv4(tos_del=tos_del.LOW, tos_thr=1).data + # ``LOW`` is a ToSDelay member but not a ToSThroughput one + self.assertEqual(ipv4(tos_del='LOW', tos_thr='LOW', tos_thr_default=1).data, expected) + self.assertEqual(ipv4(tos_thr='LOW', tos_thr_default=1).data, + ipv4(tos_thr=1).data) + + def test_non_parameter_first_argument_matches_every_argument(self) -> None: + enum = importlib.import_module('enum') + protocol = _attr('pcapkit.protocols.protocol.ProtocolBase') + level = enum.IntEnum('Level', {'HIGH': 7}) + + class Alias(protocol): # type: ignore[misc,valid-type] + def make_alias(self, b: 'Any' = level.HIGH, b_namespace: 'Any' = None) -> 'int': + value = b + return self._make_index(value) + + # ``value`` is a local, so ``b`` is found by its value as before + self.assertEqual(Alias.make_alias(Alias, b='HIGH'), 7) + + def test_unlocated_call_matches_every_argument(self) -> None: + # without the caller's source the call cannot be located, so every + # argument holding the name is a candidate and they must agree + module = importlib.import_module('pcapkit.protocols.protocol') + exc = _attr('pcapkit.utilities.exceptions.ProtocolNotImplemented') + pair, _, right = self._pair() + + with unittest.mock.patch.object(module, '_make_index_calls', lambda func: ()): + with self.assertRaisesRegex(exc, r"ambiguous member name 'Shared'.*" + r"pass a_namespace= or b_namespace= explicitly"): + pair.make_pair(pair, a='Shared', b='Shared') + self.assertEqual(pair.make_pair(pair, a='Shared', b='Shared', b_namespace=right), (1, 2)) + + def test_source_unavailable_locates_nothing(self) -> None: + module = importlib.import_module('pcapkit.protocols.protocol') + # a builtin has no Python source + self.assertEqual(module._make_index_calls(len), ()) + + +if __name__ == '__main__': + unittest.main()