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
5 changes: 5 additions & 0 deletions docs/source/pcapkit/corekit/fields/ipaddress.rst
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,11 @@ IP Interface
:members:
:show-inheritance:

Construction Helpers
~~~~~~~~~~~~~~~~~~~~

.. autofunction:: pcapkit.corekit.fields.ipaddress.parse_ip_address

Internal Definitions
~~~~~~~~~~~~~~~~~~~~

Expand Down
112 changes: 111 additions & 1 deletion pcapkit/corekit/fields/ipaddress.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,12 @@
__all__ = [
'IPv4AddressField', 'IPv6AddressField',
'IPv4InterfaceField', 'IPv6InterfaceField',
'parse_ip_address',
]

if TYPE_CHECKING:
from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface
from typing import Any, Callable, Iterator
from typing import Any, Callable, Iterator, Optional

from typing_extensions import Literal, Self

Expand Down Expand Up @@ -98,13 +99,122 @@ def _reject_bool(value: 'object', description: str) -> 'None':
wrong position does not fire, and that placement mistake has
already been made twice in this repository's history.

Guarding the field classes is necessary but not sufficient, because a
``_make_*`` that must know the address family before it can build the
schema converts the argument itself and so never hands this module a
:obj:`bool` at all. :func:`parse_ip_address` is where those callers
reach this guard (c.f. #508).

"""
if isinstance(value, bool):
raise FieldValueError(
f'{description}: must not be a bool, not {value!r} -- pass '
f'int({value!r}) if the numeric value is what is wanted')


def parse_ip_address(value: 'IPv4Address | IPv6Address | bytes | int | str',
description: str,
version: 'Optional[int]' = None) -> 'IPv4Address | IPv6Address':
"""Convert a caller-supplied address on the **construction** path.

Args:
value: Address as the caller gave it -- an :mod:`ipaddress` object,
which is returned unchanged, or anything :mod:`ipaddress` accepts.
description: Human-readable description of what ``value`` is, used to
build the :exc:`FieldValueError` message. Callers in a protocol
should carry their usual context into it, e.g.
``f'{self.alias}: [OptNo {type}] care-of address'``.
version: IP version to demand, ``4`` or ``6``, or :obj:`None` to take
whichever family ``value`` describes. Pass it where the wire format
fixes the family, so that an :class:`int` is widened to the right
one -- ``258`` is ``::102`` for ``version=6`` but ``0.0.1.2`` for
:func:`ipaddress.ip_address`.

Returns:
The converted address.

Raises:
FieldValueError: If ``value`` is a :obj:`bool` (c.f.
:func:`_reject_bool`), is not a valid IP address, or is not of
``version``.

Notes:
This is the sanctioned way for a ``_make_*`` method to turn a
caller-supplied address into an :mod:`ipaddress` object, and it exists
because doing it with :func:`ipaddress.ip_address` directly is what
#508 turned out to be: a ``_make_*`` that has to know the address
*family* before it can build the schema -- to size an option whose
length is the only thing on the wire that carries the family -- must
convert the argument itself, and that conversion happens **before** the
schema, so it launders a :obj:`bool` into an
:class:`~ipaddress.IPv4Address` that #500's guard in
:meth:`_IPAddressField.pre_process` can then only see as a legitimate
address. Seven such call sites took ``True`` / ``False`` without
complaint as ``0.0.0.1`` / ``0.0.0.0`` -- or ``::1`` / ``::`` where the
wire format fixes the family as IPv6 -- and six of them went on to pack
those octets. The seventh,
:meth:`TCP._make_mptcp_addaddr
<pcapkit.protocols.transport.tcp.TCP._make_mptcp_addaddr>`, built an
equally corrupt schema and is only stopped from packing it by an
unrelated defect of its own.

Routing every one of them through here rather than giving each its own
:func:`isinstance` check is the whole point: #481 added exactly such a
check to :meth:`MH._make_opt_mn_id
<pcapkit.protocols.internet.mh.MH._make_opt_mn_id>`, and #491 was the
same defect surviving at every site that had not been thought of. A
guard that has to be remembered per call site is a guard that will be
forgotten at the next one.

The :obj:`bool` rejection is the **first** statement here, ahead of any
dispatch on the value's type, for the placement reason #481 gives and
:func:`_reject_bool` repeats.

This raises :exc:`FieldValueError` and not
:exc:`~pcapkit.utilities.exceptions.ProtocolError`, which is deliberate
even though two sibling guards for the same mistake --
:meth:`MH._make_opt_mn_id
<pcapkit.protocols.internet.mh.MH._make_opt_mn_id>` from #481 and
:class:`ESP's SecurityAssociation
<pcapkit.protocols.internet.esp.SecurityAssociation>` from #491 -- raise
the latter. The layer decides: this is a field-level conversion, so it
answers with what :meth:`_IPAddressField.pre_process` answers with for
the identical value, and a caller sees one exception whether the
:obj:`bool` reached the field through the schema or through a
``_make_*``. The two protocol-level guards answer for the *option*,
alongside siblings that are not about addresses at all --
``_make_opt_mn_id`` refuses a :obj:`bool` for all eight MN-ID subtypes,
only one of which is address-typed -- so neither can route through here
without losing the subtype-aware message that is the point of it. Both
exception classes derive from
:exc:`~pcapkit.utilities.exceptions.BaseError` *and* :exc:`ValueError`,
so the difference is invisible to ``except BaseError`` and ``except
ValueError``, and nothing in the library catches either one
specifically.

"""
_reject_bool(value, description)

if isinstance(value, (ipaddress.IPv4Address, ipaddress.IPv6Address)):
ip = value # type: IPv4Address | IPv6Address
else:
with _reraise_as_field_value_error(description):
if version == 4:
ip = ipaddress.IPv4Address(value)
elif version == 6:
ip = ipaddress.IPv6Address(value)
else:
ip = ipaddress.ip_address(value)

# NOTE: Checked outside the ``with`` block above, and after it, because an
# :mod:`ipaddress` object taken from the branch that skips the conversion
# has not been version-checked at all -- ``IPv6Address(IPv4Address(...))``
# would have raised, but returning the object unchanged cannot.
if version is not None and ip.version != version:
raise FieldValueError(f'{description}: IP version mismatch: {ip.version} != {version}')
return ip


class _IPField(Field[_T], Generic[_T]):
"""Internal IP related value for protocol fields.

Expand Down
18 changes: 16 additions & 2 deletions pcapkit/protocols/internet/hip.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@
from pcapkit.const.hip.packet import Packet as Enum_Packet
from pcapkit.const.hip.parameter import Parameter as Enum_Parameter
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.hip import HIP as Data_HIP
from pcapkit.protocols.data.internet.hip import AckDataParameter as Data_AckDataParameter
Expand Down Expand Up @@ -3067,14 +3068,27 @@ def _make_locator(locator: 'Optional[Data_Locator]' = None, *,
preferred = locator.preferred
lifetime = math.floor(locator.lifetime.total_seconds())
else:
# NOTE: Through ``parse_ip_address`` because the locator is packed
# to octets *here*, ahead of the schema, so a bare
# ``ipaddress.IPv6Address`` would launder a ``bool`` into ``::1``
# and hand the schema's ``SwitchField`` plain bytes that its guard
# cannot question. Before this, ``ip=True`` packed a locator of
# ``::1`` with no error at all. The ``version=6`` argument is the
# *IP* version rather than the HIP one the message names, and it is
# what keeps the ``int`` widening this signature documents --
# ``0x102`` is ``::102``, not the ``0.0.1.2`` that
# ``ipaddress.ip_address`` would give (c.f. #508).
ip_val = parse_ip_address(
ip, f'HIPv{version}: [ParamNo {code}] invalid locator', version=6)

if spi is None:
length = 4
data = ipaddress.IPv6Address(ip).packed
data = ip_val.packed
else:
length = 5
data = Schema_LocatorData(
spi=spi,
ip=ipaddress.IPv6Address(ip).packed,
ip=ip_val.packed,
)

if isinstance(lifetime, timedelta):
Expand Down
60 changes: 42 additions & 18 deletions pcapkit/protocols/internet/mh.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@
UpdateNotificationACKStatus as Enum_UpdateNotificationACKStatus
from pcapkit.const.mh.upn_reason import UpdateNotificationReason as Enum_UpdateNotificationReason
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.mh import MH as Data_MH
from pcapkit.protocols.data.internet.mh import \
Expand Down Expand Up @@ -8873,8 +8874,16 @@ def _make_opt_bid(self, type: 'Enum_Option', option: 'Optional[Data_BindingIdent
if address is None:
length = 4
else:
addr = ipaddress.ip_address(address) if not isinstance(
address, (ipaddress.IPv4Address, ipaddress.IPv6Address)) else address
# NOTE: Converted through ``parse_ip_address`` rather than with
# ``ipaddress.ip_address``, because the conversion happens *here* --
# the option length below is derived from the family, so it cannot
# wait for the schema -- and a bare conversion therefore turns a
# ``bool`` into a perfectly ordinary ``IPv4Address`` that the
# schema's own guard can no longer tell from a real address. Before
# this, ``address=True`` packed as ``23080001000000000001``, i.e. a
# care-of address of ``0.0.0.1`` (c.f. #508).
addr = parse_ip_address(
address, f'{self.alias}: [OptNo {type}] invalid care-of address')
length = 8 if addr.version == 4 else 20
address = addr

Expand Down Expand Up @@ -9112,13 +9121,13 @@ def _make_opt_lmaa(self, type: 'Enum_Option', option: 'Optional[Data_LMAAddressO
# NOTE: The address is normalised rather than passed through, so that the
# schema attribute holds the same type it would after a parse. The width is
# then taken from the address itself rather than from ``code``, so that the
# two cannot be emitted disagreeing.
if isinstance(address, bytes):
addr = ipaddress.ip_address(address) # type: IPv4Address | IPv6Address
elif isinstance(address, (ipaddress.IPv4Address, ipaddress.IPv6Address)):
addr = address
else:
addr = ipaddress.ip_address(address)
# two cannot be emitted disagreeing. Normalising *here*, ahead of the
# schema, is also why the conversion goes through ``parse_ip_address``:
# ``ipaddress.ip_address(True)`` is ``0.0.0.1``, and the schema's own
# guard cannot see that it was ever a ``bool``. Before this,
# ``address=True`` packed as ``2906010000000001`` (c.f. #508).
addr = parse_ip_address(
address, f'{self.alias}: [OptNo {type}] invalid address')

return Schema_LMAAddressOption(
type=type,
Expand Down Expand Up @@ -9314,9 +9323,13 @@ def _make_fid_suboption(self, code: 'Enum_FlowIDSuboption',
address = cast('Data_TargetCareofAddressSuboption', option).address # type: Any
else:
address = kwargs.get('address', '::')
addr = address if isinstance(
address, (ipaddress.IPv4Address, ipaddress.IPv6Address)
) else ipaddress.ip_address(address)
# NOTE: Through ``parse_ip_address`` because the sub-option length
# below is derived from the family here, ahead of the schema, so a
# bare ``ipaddress.ip_address`` would launder a ``bool`` past the
# schema's guard. Before this, ``address=True`` packed as
# ``0506000000000001`` (c.f. #508).
addr = parse_ip_address(
address, f'{self.alias}: [OptNo {code}] invalid target care-of address')
return Schema_TargetCareofAddressSuboption(
type=code, length=6 if addr.version == 4 else 18, address=addr)

Expand Down Expand Up @@ -9821,9 +9834,15 @@ def _make_opt_dmnp(self, type: 'Enum_Option', option: 'Optional[Data_DelegatedMN
prefix_length = option.prefix_length
prefix = option.prefix

addr = prefix if isinstance(
prefix, (ipaddress.IPv4Address, ipaddress.IPv6Address)
) else ipaddress.ip_address(prefix)
# NOTE: Through ``parse_ip_address`` because the ``V`` flag and the width
# are both derived from the family here, ahead of the schema, so a bare
# ``ipaddress.ip_address`` would launder a ``bool`` past the schema's
# guard. Before this, ``prefix=True`` with an IPv4-valid
# ``prefix_length`` packed as ``3706801800000001``; the default
# ``prefix_length=64`` masked it behind the range check below, which is
# why #508's own sweep read this site as already guarded (c.f. #508).
addr = parse_ip_address(
prefix, f'{self.alias}: [OptNo {type}] invalid mobile network prefix')
ipv4 = addr.version == 4

if prefix_length > (32 if ipv4 else 128):
Expand Down Expand Up @@ -10092,9 +10111,14 @@ def _make_opt_lma_up(self, type: 'Enum_Option', option: 'Optional[Data_LMAUserPl
if address is None:
return Schema_LMAUserPlaneAddressOption(type=type, length=2, address=b'')

addr = address if isinstance(
address, (ipaddress.IPv4Address, ipaddress.IPv6Address)
) else ipaddress.ip_address(address)
# NOTE: Through ``parse_ip_address`` because the option length below is
# derived from the family here, ahead of the schema, so a bare
# ``ipaddress.ip_address`` would launder a ``bool`` past the schema's
# guard. Before this, ``address=True`` packed as ``3b06000000000001``.
# ``None`` is handled above and stays an absent address, which is a
# legitimate value here and not what is being rejected (c.f. #508).
addr = parse_ip_address(
address, f'{self.alias}: [OptNo {type}] invalid LMA user-plane address')

return Schema_LMAUserPlaneAddressOption(
type=type,
Expand Down
15 changes: 13 additions & 2 deletions pcapkit/protocols/transport/tcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,6 @@
"""
import collections
import datetime
import ipaddress
import math
from typing import TYPE_CHECKING, cast

Expand All @@ -50,6 +49,7 @@
from pcapkit.const.tcp.flags import Flags as Enum_Flags
from pcapkit.const.tcp.mp_tcp_option import MPTCPOption as Enum_MPTCPOption
from pcapkit.const.tcp.option import Option as Enum_Option
from pcapkit.corekit.fields.ipaddress import parse_ip_address
from pcapkit.corekit.module import ModuleDescriptor
from pcapkit.corekit.multidict import OrderedMultiDict
from pcapkit.protocols.data.transport.tcp import CC as Data_CC
Expand Down Expand Up @@ -2894,7 +2894,18 @@ def _make_mptcp_addaddr(self, subtype: 'Enum_MPTCPOption', opt: 'Optional[Data_M
addr_val = opt.addr
port = opt.port
else:
addr_val = ipaddress.ip_address(addr)
# NOTE: Through ``parse_ip_address`` because the ``version`` sub-field
# and the option length below are both derived from the family here,
# ahead of the schema, so a bare ``ipaddress.ip_address`` would launder
# a ``bool`` into an ``IPv4Address`` that the schema's own guard can no
# longer tell from a real address -- ``addr=True`` reached
# ``mptcp_add_address_selector`` as ``0.0.0.1`` with ``version=4``
# (c.f. #508). This option cannot be constructed end to end at all for
# an unrelated reason, ``KeyError: 'length'`` from
# pcapkit/protocols/schema/transport/tcp.py:790, which is why the
# corruption here was only ever visible on the schema the maker returns.
addr_val = parse_ip_address(
addr, f'{self.alias}: [OptNo {Enum_Option.Multipath_TCP}] invalid address')
version = addr_val.version

return Schema_MPTCPAddAddress(
Expand Down
Loading
Loading