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
108 changes: 97 additions & 11 deletions pcapkit/corekit/fields/ipaddress.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
"""IP address field class"""

import abc
import contextlib
import ipaddress
from typing import TYPE_CHECKING, Generic, TypeVar, cast

Expand All @@ -15,7 +16,7 @@

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

from typing_extensions import Literal, Self

Expand All @@ -28,6 +29,45 @@
_IT = TypeVar('_IT', 'IPv4Interface', 'IPv6Interface')


@contextlib.contextmanager
def _reraise_as_field_value_error(description: str) -> 'Iterator[None]':
"""Translate a bare :exc:`ValueError` from :mod:`ipaddress` into :exc:`FieldValueError`.

Every conversion in this module ultimately calls into the stdlib
:mod:`ipaddress` module, which raises a bare :exc:`ValueError` (or a
subclass of it, e.g. :exc:`~ipaddress.AddressValueError` or
:exc:`~ipaddress.NetmaskValueError`) for a malformed value. Left alone,
that exception is not an instance of
:exc:`~pcapkit.utilities.exceptions.BaseError`, unlike every other
exception this module raises -- so a caller cannot rely on
``except BaseError`` to catch a bad field value. Wrapping the conversion
in this context manager re-raises it as :exc:`FieldValueError` instead,
preserving the original message.

Args:
description: Human-readable description of the value being
converted, used to build the :exc:`FieldValueError` message.

Raises:
FieldValueError: If the code inside the ``with`` block raises
:exc:`ValueError`.

"""
try:
yield
except FieldValueError:
# NOTE: ``FieldValueError`` is itself a ``ValueError``, so without this
# clause first, a ``FieldValueError`` raised inside the ``with`` block
# (e.g. a version-mismatch check) would be caught below and re-wrapped,
# losing its original message. Callers are expected to keep such
# raises outside the ``with`` block, but this is the same ordering
# trap ``ProtocolError`` carries at ``exceptions.py``, so it is guarded
# here too rather than relied upon by convention alone.
raise
except ValueError as error:
raise FieldValueError(f'{description}: {error}') from error


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

Expand Down Expand Up @@ -64,11 +104,16 @@ def pre_process(self, value: '_AT | bytes | int | str', packet: 'dict[str, Any]'
Returns:
Processed field value.

Raises:
FieldValueError: If ``value`` is not a valid IP address, or if it
is the wrong IP version for this field.

"""
if isinstance(value, (ipaddress.IPv4Address, ipaddress.IPv6Address)):
ip = value # type: IPv4Address | IPv6Address
else:
ip = ipaddress.ip_address(value)
with _reraise_as_field_value_error('invalid IP address'):
ip = ipaddress.ip_address(value)

if ip.version != self.version:
raise FieldValueError(f'IP version mismatch: {ip.version} != {self.version}')
Expand All @@ -84,8 +129,17 @@ def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> '_AT':
Returns:
Processed field value.

Raises:
FieldValueError: If ``value`` is the wrong IP version for this
field. ``value`` cannot actually fail the underlying
:func:`ipaddress.ip_address` conversion here -- it is always
exactly 4 or 16 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.

"""
val = ipaddress.ip_address(value)
with _reraise_as_field_value_error('invalid IP address'):
val = ipaddress.ip_address(value)
if val.version != self.version:
raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}')
return val # type: ignore[return-value]
Expand Down Expand Up @@ -179,11 +233,16 @@ def pre_process(self, value: 'IPv4Interface | bytes | int | str', packet: 'dict[
Returns:
Processed field value.

Raises:
FieldValueError: If ``value`` is not a valid IP interface, or if
it is the wrong IP version for this field.

"""
if isinstance(value, ipaddress.IPv4Interface):
val = value
else:
val = ipaddress.ip_interface(value) # type: ignore[assignment]
with _reraise_as_field_value_error('invalid IP interface'):
val = ipaddress.ip_interface(value) # type: ignore[assignment]
if val.version != self.version:
raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}')

Expand All @@ -201,16 +260,27 @@ def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'IPv4Interfa
Returns:
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.

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`.

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

val = ipaddress.ip_interface(f'{ip}/{mask}')
with _reraise_as_field_value_error('invalid IPv4 interface'):
val = ipaddress.ip_interface(f'{ip}/{mask}')
if val.version != self.version:
raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}')
return val
Expand Down Expand Up @@ -247,11 +317,16 @@ def pre_process(self, value: 'IPv6Interface | bytes | int | str', packet: 'dict[
Returns:
Processed field value.

Raises:
FieldValueError: If ``value`` is not a valid IP interface, or if
it is the wrong IP version for this field.

"""
if isinstance(value, ipaddress.IPv6Interface):
val = value
else:
val = ipaddress.ip_interface(value) # type: ignore[assignment]
with _reraise_as_field_value_error('invalid IP interface'):
val = ipaddress.ip_interface(value) # type: ignore[assignment]
if val.version != self.version:
raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}')

Expand All @@ -271,21 +346,32 @@ def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'IPv6Interfa

Raises:
FieldValueError: If the trailing octet is not a valid IPv6 prefix
length, i.e. greater than 128.
length, i.e. greater than 128, or if the resulting interface
is the wrong IP version for this field. Neither the leading
sixteen octets nor the final :func:`ipaddress.ip_interface`
call can actually fail here -- the former is always exactly
16 octets, fixed by this field's length, and any such octet
string is a valid address; the latter is only ever reached
once the prefix length has already been checked above, and
any prefix length in ``0..128`` is valid. Both conversions
are still wrapped for consistency with the rest of this
module.

Notes:
The trailing octet is the prefix length as a binary integer, as
written by :meth:`pre_process` -- not a dotted netmask as in
:meth:`IPv4InterfaceField.post_process`.

"""
ip = ipaddress.IPv6Address(value[:16])
with _reraise_as_field_value_error('invalid IPv6 address'):
ip = ipaddress.IPv6Address(value[:16])
prefixlen = value[16]

if prefixlen > 128:
raise FieldValueError(f'invalid IPv6 prefix length: {prefixlen}')

val = ipaddress.ip_interface(f'{ip}/{prefixlen}')
with _reraise_as_field_value_error('invalid IPv6 interface'):
val = ipaddress.ip_interface(f'{ip}/{prefixlen}')
if val.version != self.version:
raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}')
return val
77 changes: 77 additions & 0 deletions tests/corekit/test_fields_ipaddress.py
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,83 @@ def test_ipv4_interface_rejects_the_other_version(self) -> None:
with self.assertRaises(FieldValueError):
IPv4InterfaceField().pre_process('2001:db8::1/64', {})

def test_pre_process_malformed_value_raises_in_library_error(self) -> None:
"""A malformed address/interface string used to let a bare
:exc:`ValueError` from :mod:`ipaddress` escape -- unlike the very next
statement in the same method, which already raised the library's own
:exc:`FieldValueError` for a value that is merely the wrong IP
version. So ``except BaseError`` could not reliably catch a bad field
value: whether the exception was in-library depended on *how* the
value was wrong. All four public field classes must now raise
:exc:`FieldValueError` here too, with the original :mod:`ipaddress`
message preserved.
"""
from pcapkit.corekit.fields.ipaddress import (
IPv4AddressField, IPv6AddressField, IPv4InterfaceField, IPv6InterfaceField,
)
from pcapkit.utilities.exceptions import BaseError, FieldValueError

cases = [
(IPv4AddressField(), 'not-an-address'),
(IPv6AddressField(), 'not-an-address'),
(IPv4InterfaceField(), 'not-an-interface'),
(IPv6InterfaceField(), 'not-an-interface'),
]
for field, bad_value in cases:
with self.subTest(field=type(field).__name__):
with self.assertRaises(FieldValueError) as context:
field.pre_process(bad_value, {})
# ``FieldValueError`` subclasses ``BaseError``, but assert the
# in-library type directly (above) rather than only this.
self.assertIsInstance(context.exception, BaseError)
# the original stdlib ``ipaddress`` message must survive the
# translation, not just some generic replacement text
self.assertIn(repr(bad_value), str(context.exception))

def test_wrong_version_message_is_not_relabelled_as_a_malformed_value(self) -> None:
"""Wrapping the conversion must not broaden into swallowing the
pre-existing wrong-version ``FieldValueError`` -- its message stays
the version-mismatch message, not the "invalid IP ..." message used
for a genuinely malformed value.
"""
from pcapkit.corekit.fields.ipaddress import IPv6AddressField, IPv6InterfaceField
from pcapkit.utilities.exceptions import FieldValueError

with self.assertRaises(FieldValueError) as address_context:
IPv6AddressField().pre_process(ipaddress.IPv4Address('1.2.3.4'), {})
self.assertIn('IP version mismatch', str(address_context.exception))
self.assertNotIn('invalid IP', str(address_context.exception))

with self.assertRaises(FieldValueError) as interface_context:
IPv6InterfaceField().pre_process(ipaddress.IPv4Interface('1.2.3.4/24'), {})
self.assertIn('IP version mismatch', str(interface_context.exception))
self.assertNotIn('invalid IP', str(interface_context.exception))

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.
"""
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))


if __name__ == '__main__':
unittest.main()
Loading