Skip to content

fix(ipv4,tcp)!: a header length declared past the data zero-fills on rebuild; TS make() rejects string address keys (#1402, #1404) #1411

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JarryShaw merged 1 commit into from
Oct 8, 2026
Merged

fix(ipv4,tcp)!: a header length declared past the data zero-fills on rebuild; TS make() rejects string address keys (#1402, #1404) #1411

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
31 changes: 26 additions & 5 deletions pcapkit/protocols/internet/ipv4.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@
from pcapkit.const.ipv4.tos_thr import ToSThroughput as Enum_ToSThroughput
from pcapkit.const.ipv4.ts_flag import TSFlag as Enum_TSFlag
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.corekit.fields.ipaddress import parse_ip_address
from pcapkit.corekit.multidict import OrderedMultiDict
from pcapkit.protocols.data.internet.ipv4 import EOOLOption as Data_EOOLOption
from pcapkit.protocols.data.internet.ipv4 import ESECOption as Data_ESECOption
Expand Down Expand Up @@ -283,6 +284,10 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu
Returns:
Parsed packet data.

Raises:
ProtocolError: If the version is not 4, or if ``ihl`` declares
more header than the data holds.

"""
if length is None:
length = len(self)
Expand Down Expand Up @@ -316,6 +321,13 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=unused-argu
dst=schema.dst,
)

# NOTE: An ``ihl`` declaring more header than was captured leaves the
# missing octets to read as zeros, which ``from_data`` then rebuilds;
# rejecting it keeps the octets as captured inside the parent (:issue:`1404`).
if len(self) < ipv4.hdr_len:
raise ProtocolError(f'{self.alias}: header length {ipv4.hdr_len} runs past the end of '
f'the data ({len(self)} octets)')

_optl = ipv4.hdr_len - 20
if _optl:
options = self._read_ipv4_options(_optl)
Expand Down Expand Up @@ -1606,7 +1618,7 @@ def _make_opt_lsr(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_LSROpt
def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOption]' = None, *,
counts: 'int' = 5,
overflow: 'int' = 0,
timestamp: 'Optional[list[int | timedelta] | dict[IPv4Address, int | timedelta]]' = None,
timestamp: 'Optional[list[int | timedelta] | dict[IPv4Address | str | int, int | timedelta]]' = None,
remaining: 'Optional[list[int]]' = None,
**kwargs: 'Any') -> 'Schema_TSOption':
"""Make IPv4 Timestamp (``TS``) option.
Expand All @@ -1615,14 +1627,21 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
kind: option type code
option: option data
counts: maximum number of timestamps to record
timestamp: list of timestamps, i.e. the slots before the pointer
timestamp: list of timestamps, i.e. the slots before the pointer, or
a mapping of the addresses to their timestamps; an address may
be anything :func:`~pcapkit.corekit.fields.ipaddress.parse_ip_address`
accepts for IPv4
remaining: raw 32-bit words for the slots at or beyond the pointer;
the slots left over up to ``counts`` are zero
**kwargs: arbitrary keyword arguments

Returns:
Constructed option schema.

Raises:
FieldValueError: If an address is a :obj:`bool` or is not a valid
IPv4 address.

"""
if option is not None:
ts_list = [] # type: list[int]
Expand All @@ -1638,7 +1657,8 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
ts_list.append(ts)
else:
for ip, ts in option.timestamp.items(True):
ts_list.append(int(ip))
ts_list.append(int(parse_ip_address(ip, f'{self.alias}: [OptNo {kind}] timestamp address',
version=4)))
if not isinstance(ts, int):
ts = ts // datetime.timedelta(milliseconds=1)
if ts.bit_length() > 31:
Expand Down Expand Up @@ -1676,12 +1696,13 @@ def _make_opt_ts(self, kind: 'Enum_OptionNumber', option: 'Optional[Data_TSOptio
counts = min(4, counts) # 4 is the maximum number of timestamps
length = 4 + counts * 8

for index, (ip, ts) in enumerate(timestamp.items()):
for index, (addr, ts) in enumerate(timestamp.items()):
if index >= counts:
warn(f'{self.alias}: [OptNo {kind}] too many timestamps: {len(timestamp)}', ProtocolWarning)
break

ts_list.append(int(ip))
ts_list.append(int(parse_ip_address(addr, f'{self.alias}: [OptNo {kind}] timestamp address',
version=4)))
if not isinstance(ts, int):
ts = ts // datetime.timedelta(milliseconds=1)
if ts.bit_length() > 31:
Expand Down
11 changes: 11 additions & 0 deletions pcapkit/protocols/transport/tcp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -459,6 +459,10 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_TCP':
Returns:
Parsed packet data.

Raises:
ProtocolError: If the Data Offset declares more header than the
data holds.

"""
if length is None:
length = len(self)
Expand Down Expand Up @@ -508,6 +512,13 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_TCP':
'connection': self._flags,
})

# NOTE: A Data Offset declaring more header than was captured leaves the
# missing octets to read as zeros, which ``from_data`` then rebuilds;
# rejecting it keeps the octets as captured inside the parent (:issue:`1404`).
if len(self) < tcp.hdr_len:
raise ProtocolError(f'{self.alias}: header length {tcp.hdr_len} runs past the end of '
f'the data ({len(self)} octets)')

_optl = tcp.hdr_len - 20
if _optl:
# NOTE: the octets after an End of Option List option are kept as
Expand Down
58 changes: 33 additions & 25 deletions tests/protocols/internet/test_ipv4_unit.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1981,52 +1981,60 @@ def test_ipv4_schema_helpers_and_post_process_branches(self) -> None:
self.assertEqual(tuple(unknown.timestamp), (1,))
warn.assert_called_once()

def test_an_option_area_longer_than_the_datagram_still_parses(self) -> None:
"""An ``ihl`` promising more options than are there is tolerated.
def test_an_option_area_longer_than_the_datagram_is_refused(self) -> None:
"""An ``ihl`` promising more options than are there is refused. C.f. #431, #1404.

The header below sets ``ihl`` to 10 -- a 20-octet option area -- and stops
after the fixed 20 octets, so there are no option octets at all. Reading
past them yields ``b''``, which decodes the option number as 0, and 0 is
IPv4's end-of-option-list, so the option loop breaks there and reports the
whole area as padding. That is how a datagram cut short by the snapshot
length parses at all, and it is why :meth:`OptionField.unpack
after the fixed 20 octets, so there are no option octets at all. Read by
the schema, the option area yields ``b''``, which decodes the option
number as 0, and 0 is IPv4's end-of-option-list, so the option loop breaks
there. That is why :meth:`OptionField.unpack
<pcapkit.corekit.fields.collections.OptionField.unpack>` checks each
option's progress *after* its end-of-option-list break rather than before:
checking first turns every such header into an error. C.f. #431.

:class:`~pcapkit.protocols.internet.ipv4.IPv4` itself refuses the header
with :exc:`~pcapkit.utilities.exceptions.ProtocolError`, since the 20
missing octets would otherwise be rebuilt as zeros (#1404).

"""
from pcapkit.const.ipv4.option_number import OptionNumber
from pcapkit.protocols.internet.ipv4 import IPv4
from pcapkit.protocols.schema.internet.ipv4 import IPv4 as Schema_IPv4
from pcapkit.utilities.exceptions import ProtocolError
from tests._support import time_limit

raw = bytes.fromhex('4a00001800010000400600000a0000010a000002')
with time_limit():
proto = IPv4(raw, len(raw))

self.assertEqual(proto.info.hdr_len, 40)
schema = Schema_IPv4.unpack(raw)
self.assertEqual(
[(code, opt.length) for code, opt in proto.info.options.items(multi=True)],
[(opt.type, opt.length) for opt in schema.options],
[(OptionNumber.EOOL, 1)],
)

with self.assertRaisesRegex(ProtocolError, 'header length 40 runs past the end of the data'):
IPv4(raw, len(raw))

def test_a_truncated_option_still_parses_its_declared_length(self) -> None:
"""A capture cut short mid-option is refused alone, and kept as captured in a frame. C.f. #431, #572, #1325.

:meth:`test_an_option_area_longer_than_the_datagram_still_parses` above
:meth:`test_an_option_area_longer_than_the_datagram_is_refused` above
pins the *empty*-tail half of the #431 accommodation: an option area
that runs out before it starts. This is the other half -- an option
that *does* start, declares more data than the capture holds, and runs
out partway through.

The header below sets ``ihl`` to 9 -- 16 declared octets of options --
for an unassigned option (code 31) declaring ``length=12``, with only
6 of its 10 data octets behind it. Those 4 missing octets used to be
read as zeros and rebuilt as if captured. :meth:`OptionField.unpack
<pcapkit.corekit.fields.collections.OptionField.unpack>` now refuses
the option with :exc:`~pcapkit.utilities.exceptions.ProtocolError`
instead (#1325). Inside a frame the datagram is kept as the octets
captured, a :class:`~pcapkit.protocols.misc.raw.Raw` payload carrying
that error, and rebuilds byte for byte.
The header below sets ``ihl`` to 7 -- 8 octets of options, all present,
with 8 octets of payload behind them -- for an unassigned option (code
31) declaring ``length=12``, with only 6 of its 10 data octets inside
the option area. Those 4 missing octets used to be read as zeros and
rebuilt as if captured. The header itself is complete, so the
header-length check (#1404) passes, and :meth:`OptionField.unpack
<pcapkit.corekit.fields.collections.OptionField.unpack>` refuses the
option with :exc:`~pcapkit.utilities.exceptions.ProtocolError` (#1325).
Inside a frame the datagram is kept as the octets captured, a
:class:`~pcapkit.protocols.misc.raw.Raw` payload carrying that error,
and rebuilds byte for byte.

"""
from pcapkit.const.ipv4.option_number import OptionNumber
Expand All @@ -2038,18 +2046,18 @@ def test_a_truncated_option_still_parses_its_declared_length(self) -> None:

custom = OptionNumber.get(31)
trailing = bytes.fromhex('aabbccddeeff')
raw = (bytes.fromhex('4900001c00010000400600000a0000010a000002') +
bytes([custom, 12]) + trailing)
raw = (bytes.fromhex('4700002400010000400600000a0000010a000002') +
bytes([custom, 12]) + trailing + bytes(8))
with time_limit(), warnings.catch_warnings():
warnings.simplefilter('ignore')
with self.assertRaisesRegex(ProtocolError, 'runs past the end of the data'):
with self.assertRaisesRegex(ProtocolError, 'has an option that runs past the end of the data'):
IPv4(raw, len(raw))

frame = bytes.fromhex('0123456789ab' 'fedcba987654' '0800') + raw
proto = Ethernet(frame, len(frame))

self.assertIsInstance(proto.payload, Raw)
self.assertIn('runs past the end of the data', str(proto.payload.info.error))
self.assertIn('has an option that runs past the end of the data', str(proto.payload.info.error))
# the option octets are kept as captured, with nothing synthesised
self.assertEqual(proto.payload.data, raw)
self.assertEqual(Ethernet.from_data(proto.info).data, frame)
Expand Down
35 changes: 24 additions & 11 deletions tests/protocols/test_declared_length_overrun_unit.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,10 @@
the parent rebuilds byte for byte.

An option that begins at the end of the data reads nothing at all, and is still
left to the end-of-option-list handling an over-long option area relies on.
left to the end-of-option-list handling an over-long option area relies on. An
IPv4 or TCP header declaring an option area past the data is rejected before its
options are read (GitHub issue #1404), and its parent likewise keeps it as
captured.

Every case builds its own octets in memory and reads no capture.

Expand Down Expand Up @@ -74,9 +77,10 @@ def _ipv6(proto: int, payload: bytes) -> bytes:
bytes.fromhex('01000020 00000001 00010000 00010001 00000001 000c0010 00050000'))),
# an unassigned kind declaring 12 octets, 8 present
'TCP unassigned': ('pcapkit.protocols.transport.tcp', 'TCP', _tcp(bytes.fromhex('4f0caabb ccddeeff'))),
# an unassigned option declaring 12 octets, 8 present in a 16-octet area
# an unassigned option declaring 12 octets, 8 present in a complete
# 8-octet area, with 8 octets of payload behind it
'IPv4 unassigned': ('pcapkit.protocols.internet.ipv4', 'IPv4',
_ipv4(bytes.fromhex('1f0caabb ccddeeff'), ihl=9)),
_ipv4(bytes.fromhex('1f0caabb ccddeeff'), ihl=7, payload=bytes(8))),
}

#: The same option areas with every option complete.
Expand Down Expand Up @@ -128,7 +132,7 @@ def test_an_overrunning_option_raises_protocol_error(self) -> None:
protocol = self._protocol(module, name)
with warnings.catch_warnings():
warnings.simplefilter('ignore')
with self.assertRaisesRegex(ProtocolError, 'runs past the end of the data'):
with self.assertRaisesRegex(ProtocolError, 'has an option that runs past the end of the data'):
protocol(data, len(data))

def test_a_complete_option_still_rebuilds_byte_for_byte(self) -> None:
Expand All @@ -150,24 +154,33 @@ def test_an_overrunning_layer_is_kept_as_captured_by_its_parent(self) -> None:

if name == 'IPv6':
# an extension header is kept as raw extension data
self.assertIn('runs past the end of the data', str(parsed.info.ext.error))
self.assertIn('has an option that runs past the end of the data', str(parsed.info.ext.error))
else:
# any other layer is kept as a raw payload, as captured
self.assertIsInstance(parsed.payload, Raw)
self.assertIn('runs past the end of the data', str(parsed.payload.info.error))
self.assertIn('has an option that runs past the end of the data', str(parsed.payload.info.error))
self.assertEqual(parsed.payload.data, OVERRUNS[case][2])
self.assertEqual(protocol.from_data(parsed.info).data, frame)

def test_an_option_area_declared_past_the_data_still_ends_at_eool(self) -> None:
def test_an_option_area_declared_past_the_data_is_kept_as_captured(self) -> None:
from pcapkit.protocols.internet.ipv4 import IPv4
from pcapkit.protocols.link.ethernet import Ethernet
from pcapkit.protocols.misc.raw import Raw
from pcapkit.utilities.exceptions import ProtocolError

# ``ihl`` declares 20 octets of options with none present: the first
# option begins at the end of the data, so it is end-of-option-list.
# ``ihl`` declares 20 octets of options with none present: reading the
# option area would zero-fill all 20 of them on rebuild (issue #1404).
data = bytes.fromhex('4a00001800010000400600000a0000010a000002')
with self.assertRaisesRegex(ProtocolError, 'header length 40 runs past the end of the data'):
IPv4(data, len(data))

frame = bytes.fromhex('0123456789ab fedcba987654 0800') + data
with warnings.catch_warnings():
warnings.simplefilter('ignore')
parsed = IPv4(data, len(data))
self.assertEqual(parsed.info.src.compressed, '10.0.0.1')
parsed = Ethernet(frame, len(frame))
self.assertIsInstance(parsed.payload, Raw)
self.assertEqual(parsed.payload.data, data)
self.assertEqual(Ethernet.from_data(parsed.info).data, frame)


if __name__ == '__main__':
Expand Down
Loading
Loading