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
14 changes: 10 additions & 4 deletions pcapkit/protocols/protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -727,19 +727,25 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> '_PT':

Raises:
ProtocolError: If the made header holds raw items and its packed
octets do not parse.
octets do not parse, either as a schema or by :meth:`read`.
The message carries the original error's, which is also
chained as ``__cause__``.

"""
packet = kwargs.get('__packet__', {}) # packet data
if cast('Optional[_ST]', self.__header__) is None:
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__):
# NOTE: A schema may accept octets that :meth:`read` then rejects,
# e.g. an option whose length runs past the option area, so the
# read is covered too (:issue:`1394`).
try:
parsed = cast('_ST', type(self.__header__).unpack(io.BytesIO(self._data), length, packet)) # type: ignore[call-arg,misc]
if _raw_where_parsed(self.__header__, parsed):
self.__header__ = parsed
return self.read(length, **kwargs)
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
raise ProtocolError(f'{type(self).__name__}: malformed raw item given to make(): {error}') from error
return self.read(length, **kwargs)

@staticmethod
Expand Down
96 changes: 96 additions & 0 deletions tests/protocols/test_make_raw_item_read_unit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,96 @@
# -*- coding: utf-8 -*-
"""A raw ``make()`` item rejected by ``read`` raises the wrapped error.

GitHub issue #1394: :meth:`~pcapkit.protocols.protocol.ProtocolBase.unpack`
wrapped only the schema re-parse of a made header holding raw items. A raw
item the schema accepts but ``read`` rejects -- an option whose length runs
past the option area, or one cut short of its own fields -- escaped as the
reader's bare :exc:`~pcapkit.utilities.exceptions.ProtocolError`. The read is
now covered too: the error says ``malformed raw item given to make()``,
carries the reader's message, and chains it as ``__cause__``.

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 warnings
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 that the schema parses and ``read`` rejects:
#: (protocol class, construction keywords, the reader's own message).
READ_REJECTED = {
# the lone ``0x02`` takes the next item's kind octet as its length, 148,
# which runs past the 8-octet option area
'IPv4 overrun': ('pcapkit.protocols.internet.ipv4.IPv4',
{'options': [b'\x02', b'\x94\x04\x00\x00'], 'protocol': 253},
r'IPv4: invalid format'),
'IPv4 short': ('pcapkit.protocols.internet.ipv4.IPv4',
{'options': [b'\x0c'], 'protocol': 253},
r'IPv4: \[OptNo 12\] invalid format'),
'TCP': ('pcapkit.protocols.transport.tcp.TCP',
{'options': [b'\x03']},
r'TCP: \[OptNo 3\] invalid format'),
'HOPOPT': ('pcapkit.protocols.internet.hopopt.HOPOPT',
{'extension': True, 'options': [b'\x23']},
r'HOPOPT: \[OptNo 35\] invalid format'),
'IPv6_Opts': ('pcapkit.protocols.internet.ipv6_opts.IPv6_Opts',
{'extension': True, 'options': [b'\x26']},
r'IPv6-Opts: \[OptNo 38\] invalid format'),
'MH': ('pcapkit.protocols.internet.mh.MH',
{'data': {'options': [b'\x2b']}},
r'MH: \[Opt 43\] invalid format'),
'SCTP': ('pcapkit.protocols.transport.sctp.SCTP',
{'chunks': [b'\x01']},
r'SCTP: \[Chunk 1\] invalid format'),
# ESP_INFO needs a 12-octet body, and this one has 8
'HIP': ('pcapkit.protocols.internet.hip.HIP',
{'extension': True, 'parameters': [b'\x00\x41\x00\x08' + bytes(8)]},
r'HIPv6: \[ParamNo 65\] invalid format'),
}


class TestRawItemRejectedByRead(unittest.TestCase):
"""A raw item ``read`` rejects raises the wrapped ``ProtocolError`` (#1394)."""

def setUp(self) -> None:
reimport_once_per_class(self)

def test_read_rejection_is_wrapped(self) -> None:
exc = _attr('pcapkit.utilities.exceptions.ProtocolError')
for case, (klass, kwargs, message) in READ_REJECTED.items():
with self.subTest(case=case):
cls = _attr(klass)
with warnings.catch_warnings():
warnings.simplefilter('ignore')
with self.assertRaisesRegex(exc, r'malformed raw item given to make\(\)') as caught:
cls(**kwargs)
# the reader's own error is chained, and its message kept
self.assertIsInstance(caught.exception.__cause__, exc)
self.assertRegex(str(caught.exception.__cause__), message)
self.assertRegex(str(caught.exception), message)

def test_well_formed_raw_item_still_builds(self) -> None:
# the wider ``try`` leaves a raw item that does parse alone
ipv4 = _attr('pcapkit.protocols.internet.ipv4.IPv4')
proto = ipv4(options=[b'\x94\x04\x00\x00'], protocol=253)
self.assertEqual(proto.data.hex(), '460000180000000000fd00007f0000010000000094040000')


if __name__ == '__main__':
unittest.main()
Loading