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
38 changes: 33 additions & 5 deletions pcapkit/corekit/fields/strings.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

from pcapkit.corekit.fields.field import Field, NoValue
from pcapkit.utilities.compat import Dict
from pcapkit.utilities.exceptions import FieldValueError

__all__ = [
'_TextField',
Expand Down Expand Up @@ -196,6 +197,19 @@ def __init__(self, length: 'int',

self._namespace = namespace or {}

# NOTE: A subfield reaching past the end of the field is a mistake in the
# schema that declares it, not something a packet can cause, so it is
# rejected here rather than at packing time -- where it used to grow the
# buffer instead, and surface much later as an opaque ``OverflowError``
# from :meth:`int.to_bytes`.
width = self.length * 8
for name, (start, size) in self._namespace.items():
if start < 0 or size < 1 or start + size > width:
raise FieldValueError(
f'{type(self).__name__}: subfield {name!r} spans bits '
f'{start}-{start + size - 1} of a {width}-bit field'
)

def pre_process(self, value: 'dict[str, Any]', packet: 'dict[str, Any]') -> 'bytes': # pylint: disable=unused-argument
"""Process field value before construction (packing).

Expand All @@ -206,12 +220,26 @@ def pre_process(self, value: 'dict[str, Any]', packet: 'dict[str, Any]') -> 'byt
Returns:
Processed field value.

Raises:
FieldValueError: If a subfield value does not fit the bits it was
declared to occupy.

"""
buffer = bytearray(self.length * 8)
for name, (start, len) in self._namespace.items():
end = start + len
buffer[start:end] = f'{value[name]:0{end - start}b}'.encode()
return int(b''.join(map(lambda x: b'1' if x else b'0', buffer)), 2).to_bytes(self.length, 'big')
# NOTE: The buffer holds one ASCII digit per bit, so it must be seeded with
# ``b'0'`` rather than with NUL bytes -- a zero bit written by the loop below
# is the *character* ``b'0'``, which is itself non-zero, so a truthiness test
# over the buffer cannot tell a cleared bit from a set one and would report
# every named bit as set.
buffer = bytearray(b'0' * (self.length * 8))
for name, (start, size) in self._namespace.items():
bits = f'{value[name]:0{size}b}'
if len(bits) > size:
raise FieldValueError(
f'{type(self).__name__}: subfield {name!r} value {value[name]!r} '
f'needs {len(bits)} bits, but was declared with {size}'
)
buffer[start:start + size] = bits.encode()
return int(buffer, 2).to_bytes(self.length, 'big')
Comment thread
Copilot marked this conversation as resolved.

def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'dict[str, Any]': # pylint: disable=unused-argument
"""Process field value after parsing (unpacked).
Expand Down
2 changes: 1 addition & 1 deletion pcapkit/protocols/schema/internet/ipv4.py
Original file line number Diff line number Diff line change
Expand Up @@ -526,7 +526,7 @@ class IPv4(Schema):
#: Version and header length.
vihl: 'VerIHLField' = BitField(length=1, namespace={
'version': (0, 4),
'ihl': (4, 8),
'ihl': (4, 4),
})
#: Type of service.
tos: 'ToSField' = BitField(length=1, namespace={
Expand Down
104 changes: 104 additions & 0 deletions tests/corekit/test_fields_strings.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,104 @@
from __future__ import annotations

import unittest

from tests._support import purge_modules


class BitFieldTests(unittest.TestCase):
"""Packing and parsing of :class:`~pcapkit.corekit.fields.strings.BitField`."""

def setUp(self) -> None:
purge_modules(['pcapkit'])

from pcapkit.corekit.fields.strings import BitField

self.BitField = BitField
#: One octet of single-bit flags, as the protocol schemas declare them.
self.flags = BitField(length=1, namespace={
'a': (0, 1),
'b': (1, 1),
'c': (6, 1),
'd': (7, 1),
})

def test_cleared_bits_stay_cleared_when_a_sibling_is_set(self) -> None:
"""A zero bit must not be packed as a one.

The buffer holds one ASCII digit per bit, and ``b'0'`` is itself a
non-zero byte, so a truthiness test over it reported every named bit as
set regardless of its value.

"""
self.assertEqual(self.flags.pre_process({'a': 0, 'b': 0, 'c': 0, 'd': 1}, {}), b'\x01')
self.assertEqual(self.flags.pre_process({'a': 1, 'b': 0, 'c': 0, 'd': 0}, {}), b'\x80')
self.assertEqual(self.flags.pre_process({'a': 0, 'b': 1, 'c': 1, 'd': 0}, {}), b'\x42')

def test_no_bits_set_packs_to_zero(self) -> None:
self.assertEqual(self.flags.pre_process({'a': 0, 'b': 0, 'c': 0, 'd': 0}, {}), b'\x00')

def test_every_bit_set_packs_to_its_mask(self) -> None:
self.assertEqual(self.flags.pre_process({'a': 1, 'b': 1, 'c': 1, 'd': 1}, {}), b'\xc3')

def test_multi_bit_subfields_keep_their_own_width(self) -> None:
field = self.BitField(length=1, namespace={'kind': (0, 3), 'value': (3, 5)})

self.assertEqual(field.pre_process({'kind': 5, 'value': 17}, {}), b'\xb1')
self.assertEqual(field.pre_process({'kind': 0, 'value': 1}, {}), b'\x01')

def test_unnamed_bits_are_left_alone(self) -> None:
"""Bits outside the namespace belong to nobody and must pack as zero."""
field = self.BitField(length=2, namespace={'flag': (3, 1)})

self.assertEqual(field.pre_process({'flag': 1}, {}), b'\x10\x00')
self.assertEqual(field.pre_process({'flag': 0}, {}), b'\x00\x00')

def test_a_subfield_reaching_past_the_field_is_rejected_at_declaration(self) -> None:
"""A namespace entry wider than the field is a schema mistake.

Slice assignment on a :class:`bytearray` *grows* it, so declaring
``ihl: (4, 8)`` on a one-octet field built a 12-bit buffer and then
failed in :meth:`int.to_bytes` with an opaque ``OverflowError``, far
from the declaration at fault. ``pcapkit/protocols/schema/internet/
ipv4.py`` really did carry that declaration.

"""
from pcapkit.utilities.exceptions import FieldValueError

with self.assertRaises(FieldValueError):
self.BitField(length=1, namespace={'version': (0, 4), 'ihl': (4, 8)})
with self.assertRaises(FieldValueError):
self.BitField(length=1, namespace={'flag': (8, 1)})
with self.assertRaises(FieldValueError):
self.BitField(length=1, namespace={'flag': (-1, 1)})

def test_a_value_too_wide_for_its_subfield_is_rejected(self) -> None:
from pcapkit.utilities.exceptions import FieldValueError

field = self.BitField(length=1, namespace={'kind': (0, 3), 'value': (3, 5)})

with self.assertRaises(FieldValueError):
field.pre_process({'kind': 8, 'value': 0}, {})
self.assertEqual(field.pre_process({'kind': 7, 'value': 31}, {}), b'\xff')

def test_ipv4_version_and_ihl_pack_into_one_octet(self) -> None:
"""The real declaration this guard was written for."""
from pcapkit.protocols.schema.internet.ipv4 import IPv4

field = IPv4.__fields__['vihl']

self.assertEqual(field.pre_process({'version': 4, 'ihl': 5}, {}), b'\x45')
self.assertEqual(field.post_process(b'\x45', {}), {'version': 4, 'ihl': 5})

def test_pack_and_parse_round_trip(self) -> None:
for value in ({'a': 0, 'b': 0, 'c': 0, 'd': 0},
{'a': 0, 'b': 0, 'c': 0, 'd': 1},
{'a': 1, 'b': 0, 'c': 1, 'd': 0},
{'a': 1, 'b': 1, 'c': 1, 'd': 1}):
with self.subTest(value=value):
packed = self.flags.pre_process(value, {})
self.assertEqual(self.flags.post_process(packed, {}), value)


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