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
27 changes: 27 additions & 0 deletions pcapkit/corekit/fields/numbers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down Expand Up @@ -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
Expand All @@ -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':
Expand Down
14 changes: 13 additions & 1 deletion pcapkit/protocols/schema/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
83 changes: 83 additions & 0 deletions tests/protocols/transport/test_tcp_udp_schema_port_range_unit.py
Original file line number Diff line number Diff line change
@@ -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()
Loading