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
225 changes: 214 additions & 11 deletions pcapkit/corekit/fields/ipaddress.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import abc
import contextlib
import ipaddress
import re
from typing import TYPE_CHECKING, Generic, TypeVar, cast

from pcapkit.corekit.fields.field import NO_VALUE, Field
Expand Down Expand Up @@ -347,6 +348,191 @@ class _IPInterfaceField(_IPField[_IT]):
"""


class _RawMaskIPv4Interface(ipaddress.IPv4Interface):
"""IPv4 interface whose mask is not a contiguous netmask.

Args:
address: Dotted address and mask, as ``'a.b.c.d/m.m.m.m'`` or
``'a.b.c.d/0xXXXXXXXX'``; either way the mask is read literally.

:class:`ipaddress.IPv4Interface` can only hold a prefix length, so it
reads a hostmask-shaped mask such as ``0.0.0.255`` as ``/24`` and rejects
a mask such as ``255.0.255.0``. This class keeps the mask octets as given
in :attr:`netmask`, so :meth:`IPv4InterfaceField.pre_process` writes them
back unchanged.

:func:`str`, :func:`repr` and :mod:`pickle` use the ``address/mask``
form, which :meth:`IPv4InterfaceField.pre_process` packs back to the same
octets. The mask is dotted, except when it is hostmask-shaped (ones, then
zeros, read from the low bit): :func:`ipaddress.ip_interface` would read
such a dotted mask as a hostmask, so it is written as eight hex digits
instead, e.g. ``10.0.0.1/0x000000ff``. The stdlib rejects that hex form,
so it never means anything else there.

Equality and ordering use the mask too. An instance sorts as the stdlib
sorts ``address/32``, then by mask, so the order stays total when mixed
with :class:`ipaddress.IPv4Interface` values.

Such a mask has no prefix length, so :attr:`network`, :attr:`with_prefixlen`
and the prefix length all describe the address alone (``/32``); only
:attr:`netmask`, :attr:`hostmask`, :attr:`with_netmask` and
:attr:`with_hostmask` carry the mask. :attr:`with_netmask` is always
dotted, so it is not the round-trip form; :func:`str` is.

"""

def __init__(self, address: 'str') -> 'None':
addr, _, mask = address.partition('/')
super().__init__(addr)
self.netmask = _parse_literal_mask(mask)

@property
def hostmask(self) -> 'IPv4Address':
"""Bitwise inverse of :attr:`netmask`."""
return ipaddress.IPv4Address(int(self.netmask) ^ 0xFFFFFFFF)

def __str__(self) -> 'str':
if _is_hostmask(self.netmask):
return f'{self.ip}/0x{int(self.netmask):08x}'
return f'{self.ip}/{self.netmask}'

def __eq__(self, other: 'object') -> 'bool':
if not isinstance(other, ipaddress.IPv4Interface):
return NotImplemented
return self.ip == other.ip and self.netmask == other.netmask

def __hash__(self) -> 'int':
return hash((int(self.ip), int(self.netmask)))

def __lt__(self, other: 'Any') -> 'bool':
if not isinstance(other, ipaddress.IPv4Interface):
return ipaddress.IPv4Interface.__lt__(self, other)
return _interface_order(self) < _interface_order(other)

def __le__(self, other: 'Any') -> 'bool':
if not isinstance(other, ipaddress.IPv4Interface):
return ipaddress.IPv4Interface.__le__(self, other)
return _interface_order(self) <= _interface_order(other)

def __gt__(self, other: 'Any') -> 'bool':
if not isinstance(other, ipaddress.IPv4Interface):
return ipaddress.IPv4Interface.__gt__(self, other)
return _interface_order(self) > _interface_order(other)

def __ge__(self, other: 'Any') -> 'bool':
if not isinstance(other, ipaddress.IPv4Interface):
return ipaddress.IPv4Interface.__ge__(self, other)
return _interface_order(self) >= _interface_order(other)


def _is_netmask(mask: 'IPv4Address') -> 'bool':
"""Tell whether ``mask`` is a contiguous netmask (leading ones, then zeros).

Args:
mask: Dotted mask.

Returns:
:data:`True` for ``0.0.0.0`` through ``255.255.255.255`` with no gap
in the ones, :data:`False` otherwise.

"""
inverse = int(mask) ^ 0xFFFFFFFF
return not inverse & (inverse + 1)


def _is_hostmask(mask: 'IPv4Address') -> 'bool':
"""Tell whether ``mask`` is hostmask-shaped (leading zeros, then ones).

Args:
mask: Dotted mask.

Returns:
:data:`True` when :func:`ipaddress.ip_interface` would accept the
dotted ``mask`` as a hostmask, :data:`False` otherwise.

"""
value = int(mask)
return not value & (value + 1)


#: Hex spelling of a literal mask, as :func:`str` writes a hostmask-shaped one.
_HEX_MASK = re.compile(r'0[xX][0-9a-fA-F]{8}')


def _parse_literal_mask(text: 'str') -> 'IPv4Address':
"""Read a mask literally, from eight hex digits or a dotted quad.

Args:
text: ``0xXXXXXXXX`` or ``m.m.m.m``.

Returns:
The mask as given, never reinterpreted as a hostmask.

Raises:
ValueError: If ``text`` is neither form.

"""
if _HEX_MASK.fullmatch(text) is not None:
return ipaddress.IPv4Address(int(text, 16))
return ipaddress.IPv4Address(text)


def _interface_from_mask(ip: 'IPv4Address', mask: 'IPv4Address') -> 'IPv4Interface':
"""Build the interface for an address and a literal mask.

Args:
ip: Interface address.
mask: Mask, taken as given.

Returns:
A stdlib :class:`~ipaddress.IPv4Interface` when ``mask`` is a
contiguous netmask, otherwise a :class:`_RawMaskIPv4Interface`.

"""
if _is_netmask(mask):
return ipaddress.IPv4Interface(f'{ip}/{mask}')
return _RawMaskIPv4Interface(f'{ip}/{mask}')


def _literal_interface(value: 'str') -> 'Optional[IPv4Interface]':
"""Read an ``address/mask`` string with the mask taken literally.

Args:
value: Interface string that :func:`ipaddress.ip_interface` rejected.

Returns:
The interface, or :obj:`None` if ``value`` is not an IPv4 address
followed by a dotted or ``0xXXXXXXXX`` mask.

"""
addr, sep, mask = value.partition('/')
if not sep or ('.' not in mask and _HEX_MASK.fullmatch(mask) is None):
return None
try:
return _interface_from_mask(ipaddress.IPv4Address(addr), _parse_literal_mask(mask))
except ValueError:
return None


def _interface_order(value: 'IPv4Interface') -> 'tuple[int, int, int, int]':
"""Sort key for an IPv4 interface, consistent with ``==``.

Args:
value: IPv4 interface, either from the stdlib or a
:class:`_RawMaskIPv4Interface`.

Returns:
The stdlib order (network address, netmask, address), with a third
element that is ``0`` for a stdlib interface and ``1 + mask`` for a
:class:`_RawMaskIPv4Interface`, whose stdlib network is
``address/32``.

"""
if isinstance(value, _RawMaskIPv4Interface):
return int(value.ip), 0xFFFFFFFF, 1 + int(value.netmask), int(value.ip)
return int(value.network.network_address), int(value.network.netmask), 0, int(value.ip)


class IPv4InterfaceField(_IPInterfaceField[ipaddress.IPv4Interface]):
"""IPv4 interface value for protocol fields.

Expand Down Expand Up @@ -383,14 +569,28 @@ def pre_process(self, value: 'IPv4Interface | bytes | int | str', packet: 'dict[
:func:`_reject_bool`), is not a valid IP interface, or is the
wrong IP version for this field.

Notes:
A string means what :func:`ipaddress.ip_interface` says it
means, so ``'10.0.0.1/0.0.0.255'`` is the hostmask spelling of
``/24``. Only a string the stdlib rejects is read with its mask
taken literally: a dotted mask that is neither a netmask nor a
hostmask, such as ``'10.0.0.1/255.0.255.0'``, or eight hex
digits, such as ``'10.0.0.1/0x000000ff'``. Those are the forms
:func:`str` gives a :class:`_RawMaskIPv4Interface`.

"""
_reject_bool(value, 'invalid IP interface')

if isinstance(value, ipaddress.IPv4Interface):
val = value
else:
with _reraise_as_field_value_error('invalid IP interface'):
parsed = ipaddress.ip_interface(value)
try:
parsed = ipaddress.ip_interface(value) # type: IPv4Interface | IPv6Interface
except ValueError as error:
literal = _literal_interface(value) if isinstance(value, str) else None
if literal is None:
raise FieldValueError(f'invalid IP interface: {error}') from error
parsed = literal
if not isinstance(parsed, ipaddress.IPv4Interface):
raise FieldValueError(f'IP version mismatch: {parsed.version} != {self.version}')
val = parsed
Expand All @@ -410,26 +610,29 @@ def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'IPv4Interfa
Processed field value.

Raises:
FieldValueError: If the trailing four octets are not a valid
dotted netmask, or if the resulting interface is the wrong IP
version for this field. The leading four octets cannot
actually fail here -- they are always exactly 4 octets, fixed
by this field's length, and any such octet string is a valid
address -- but the conversion is still wrapped for
consistency with the rest of this module.
FieldValueError: If the resulting interface is the wrong IP
version for this field. The eight octets cannot actually fail
to convert here -- they are always exactly 8 octets, fixed by
this field's length, any four octets are a valid address, and
a mask that is not a netmask is kept as given -- but the
conversion is still wrapped for consistency with the rest of
this module.

Notes:
The trailing four octets are a dotted netmask, as written by
:meth:`pre_process` -- not a prefix length as in
:meth:`IPv6InterfaceField.post_process`.
:meth:`IPv6InterfaceField.post_process`. When they are not a
contiguous netmask, the value is a :class:`_RawMaskIPv4Interface`
that keeps them as given, rather than an interface the stdlib
would reinterpret (``0.0.0.255`` as the hostmask of ``/24``).

"""
with _reraise_as_field_value_error('invalid IPv4 address'):
ip = ipaddress.IPv4Address(value[:4])
mask = ipaddress.IPv4Address(value[4:])

with _reraise_as_field_value_error('invalid IPv4 interface'):
val = ipaddress.ip_interface(f'{ip}/{mask}')
val = _interface_from_mask(ip, mask)
if not isinstance(val, ipaddress.IPv4Interface):
raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}')
return val
Expand Down
12 changes: 12 additions & 0 deletions pcapkit/corekit/multidict.py
Original file line number Diff line number Diff line change
Expand Up @@ -497,6 +497,18 @@ def __eq__(self, other: 'Any') -> 'bool':

__hash__ = None # type: ignore[assignment]

def deepcopy(self, memo: 'Optional[dict]' = None) -> 'OrderedMultiDict[_KT, _VT]':
"""Return a deep copy of this object.

Notes:
:meth:`MultiDict.deepcopy` rebuilds from :meth:`to_dict` with
``flat=False``, which groups the values by key. This rebuilds from
``items(multi=True)`` instead, so the copy keeps the insertion
order of interleaved keys. :meth:`__deepcopy__` reaches it too.

"""
return self.__class__(copy.deepcopy(list(self.items(multi=True)), memo=memo))

def __reduce_ex__(self, protocol: 'SupportsIndex') -> 'tuple[type, tuple[list[tuple[_KT, _VT]]]]':
return type(self), (list(self.items(multi=True)),)

Expand Down
30 changes: 12 additions & 18 deletions tests/corekit/test_fields_ipaddress.py
Original file line number Diff line number Diff line change
Expand Up @@ -606,30 +606,24 @@ def test_switch_backed_address_makers_reject_a_bool(self) -> None:
locator_set=[{'ip': int(True)}]).pack().hex(),
'00c1001800000400000000000000000000000000000000000000000100000000')

def test_ipv4_interface_post_process_rejects_a_non_contiguous_netmask(self) -> None:
"""``IPv4InterfaceField.post_process`` builds
``ipaddress.ip_interface(f'{ip}/{mask}')`` from wire bytes whose
trailing four octets are meant to be a dotted netmask. Unlike the
leading four octets (always exactly 4 octets, so always a valid
address), those trailing octets are not guaranteed to form a
*contiguous* netmask -- e.g. a capture with a malformed or corrupted
IPv4 interface option can carry ``0.255.0.255``, which
:func:`ipaddress.ip_interface` rejects with a bare
:exc:`~ipaddress.NetmaskValueError`. This is reachable through
:meth:`~pcapkit.corekit.fields.field.FieldBase.unpack` alone, with no
malformed-length input required, and must raise :exc:`FieldValueError`
instead.
def test_ipv4_interface_post_process_keeps_a_non_contiguous_netmask(self) -> None:
"""``IPv4InterfaceField.post_process`` reads wire bytes whose trailing
four octets are meant to be a dotted netmask. Those octets are not
guaranteed to form a *contiguous* netmask -- e.g. a capture with a
malformed or corrupted IPv4 interface option can carry
``0.255.0.255``, which :func:`ipaddress.ip_interface` rejects. The
field keeps such a mask as given, so the value packs back to the same
octets (c.f. :issue:`1294`).
"""
from pcapkit.corekit.fields.ipaddress import IPv4InterfaceField
from pcapkit.utilities.exceptions import BaseError, FieldValueError

field = IPv4InterfaceField()
raw = ipaddress.IPv4Address('1.2.3.4').packed + bytes([0, 255, 0, 255])

with self.assertRaises(FieldValueError) as context:
field.unpack(raw, {})
self.assertIsInstance(context.exception, BaseError)
self.assertIn('0.255.0.255', str(context.exception))
value = field.unpack(raw, {})
self.assertIsInstance(value, ipaddress.IPv4Interface)
self.assertEqual(value.netmask, ipaddress.IPv4Address('0.255.0.255'))
self.assertEqual(field.pack(value, {}), raw)


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