diff --git a/pcapkit/corekit/fields/numbers.py b/pcapkit/corekit/fields/numbers.py index 83a81540b4..81dee09dd7 100644 --- a/pcapkit/corekit/fields/numbers.py +++ b/pcapkit/corekit/fields/numbers.py @@ -158,6 +158,14 @@ class NumberField(Field[int], Generic[_T]): -- see GitHub issue :issue:`545`. Omitting it, or passing the sign the class already fixes, stays legal. + Packing does not range-check a value: :meth:`pre_process` masks it to + :attr:`bit_length` bits, so e.g. ``70000`` in a two-octet field packs + as ``0x1170`` and ``-1`` as ``0xffff``, with no error or warning. This + is the documented contract a parsed value's round trip relies on + (:issue:`1488`). A range check belongs in the protocol's ``make`` or in + a field whose values have a range of their own, such as + :class:`PortEnumField` (:issue:`2314`). + """ __length__ = None # type: Optional[int] @@ -1065,6 +1073,20 @@ def pre_process(self, value: 'int | AppType', packet: 'dict[str, Any]') -> 'int Returns: Processed field value. + Raises: + FieldValueError: If ``value`` is a port outside this field's width, + i.e. ``0``-``65535`` for the two-octet TCP, UDP and SCTP + ports. + + Notes: + :meth:`NumberField.pre_process` masks a value to the field's width + rather than range-checking it, so without this check a schema built + directly would write port ``70000`` as ``0x1170`` and ``-1`` as + ``0xffff``. A port is a whole field with no bits to carry outside + its width, so the check costs no wire value its round trip: every + port :meth:`post_process` can return is in range. The bound is the + one :meth:`post_process` tests. See GitHub issue :issue:`2314`. + """ # NOTE: imported here rather than at module level, as # :mod:`pcapkit.const` itself imports :mod:`pcapkit.corekit`; and tested @@ -1073,6 +1095,11 @@ def pre_process(self, value: 'int | AppType', packet: 'dict[str, Any]') -> 'int if isinstance(value, AppType): value = value.port + if isinstance(value, int) and not 0 <= value < (1 << (8 * self.length)): + raise FieldValueError( + f'Field {self.name} has invalid value: port {value!r} is outside ' + f'0-{(1 << (8 * self.length)) - 1}.' + ) return super().pre_process(value, packet) def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'AppType': diff --git a/pcapkit/protocols/schema/schema.py b/pcapkit/protocols/schema/schema.py index dd15fcefd1..0bbf3e1045 100644 --- a/pcapkit/protocols/schema/schema.py +++ b/pcapkit/protocols/schema/schema.py @@ -370,7 +370,19 @@ def __new__(cls, name: 'str', bases: 'tuple[type, ...]', attrs: 'dict[str, Any]' class Schema(Mapping[str, _VT], Generic[_VT], metaclass=SchemaMeta): - """Schema for protocol headers.""" + """Schema for protocol headers. + + Note: + Building a schema directly does not range-check its field values. A + :class:`~pcapkit.corekit.fields.numbers.NumberField` masks an + out-of-range integer to its width on packing, so e.g. ``70000`` in a + two-octet field packs as ``0x1170``. The range checks live in each + protocol's :meth:`~pcapkit.protocols.protocol.Protocol.make`, and in a + field whose values have a range of their own, such as + :class:`~pcapkit.corekit.fields.numbers.PortEnumField`. See GitHub issue + :issue:`2314`. + + """ if TYPE_CHECKING: #: Mapping of name conflicts with builtin methods (original names to diff --git a/tests/protocols/transport/test_tcp_udp_schema_port_range_unit.py b/tests/protocols/transport/test_tcp_udp_schema_port_range_unit.py new file mode 100644 index 0000000000..db9032388f --- /dev/null +++ b/tests/protocols/transport/test_tcp_udp_schema_port_range_unit.py @@ -0,0 +1,83 @@ +# -*- coding: utf-8 -*- +"""A TCP or UDP schema built directly rejects a port outside ``0``-``65535``. + +GitHub issue #2314. :meth:`~pcapkit.corekit.fields.numbers.NumberField.pre_process` +masks a value to the field's width, which is the documented schema contract +(#1488), so ``Schema_UDP(srcport=70000, ...)`` packed ``0x1170`` with no error. +The owner ruled that the masking stays and that the ports get their range check +from :class:`~pcapkit.corekit.fields.numbers.PortEnumField`, the field the +schemas carry them on. These tests pin that an out-of-range port is rejected on +both ports of both schemas, and that the boundary ports still close byte-exact. + +""" +import unittest +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): a hand-picked round trip over the ``int`` boundaries of a +#: two-octet field. +__tier__ = 'ALL' + +#: Ports the issue measured masking, one past each end and well past the top. +OUT_OF_RANGE = (-1, 65536, 70000) + +#: The two ends of a two-octet port. +BOUNDARIES = (0, 65535) + + +class SchemaPortRangeTests(unittest.TestCase): + """Out-of-range ports, and the boundary round trip, on the TCP and UDP schemas.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _udp(**ports: 'int') -> 'Any': + from pcapkit.protocols.schema.transport.udp import UDP + + values = {'srcport': 1024, 'dstport': 53, 'len': 8, 'checksum': b'\x00\x00', 'payload': b''} + values.update(ports) + return UDP(**values) + + @staticmethod + def _tcp(**ports: 'int') -> 'Any': + from pcapkit.protocols.schema.transport.tcp import TCP + + flags = dict.fromkeys(('cwr', 'ece', 'urg', 'ack', 'psh', 'rst', 'syn', 'fin'), 0) + values = {'srcport': 1024, 'dstport': 80, 'seq': 0, 'ack': 0, + 'offset': {'offset': 5, 'reserved': 0, 'ns': 0}, 'flags': flags, + 'window': 0, 'checksum': b'\x00\x00', 'urgent': 0, 'options': [], 'payload': b''} + values.update(ports) + return TCP(**values) + + def test_an_out_of_range_port_is_rejected_on_packing(self) -> None: + from pcapkit.utilities.exceptions import FieldValueError + + for name, build in (('UDP', self._udp), ('TCP', self._tcp)): + for field in ('srcport', 'dstport'): + for port in OUT_OF_RANGE: + with self.subTest(schema=name, field=field, port=port): + with self.assertRaisesRegex(FieldValueError, rf'port {port} is outside 0-65535'): + build(**{field: port}).pack() + + def test_the_boundary_ports_round_trip_byte_exact(self) -> None: + from pcapkit.protocols.schema.transport.tcp import TCP + from pcapkit.protocols.schema.transport.udp import UDP + + for name, schema, build in (('UDP', UDP, self._udp), ('TCP', TCP, self._tcp)): + for srcport in BOUNDARIES: + for dstport in BOUNDARIES: + with self.subTest(schema=name, srcport=srcport, dstport=dstport): + octets = build(srcport=srcport, dstport=dstport).pack() + self.assertEqual(octets[:4], srcport.to_bytes(2, 'big') + dstport.to_bytes(2, 'big')) + parsed = schema.unpack(octets) + self.assertEqual((parsed.srcport.port, parsed.dstport.port), (srcport, dstport)) + self.assertEqual(parsed.pack(), octets) + + +if __name__ == '__main__': + unittest.main()