fix(corekit): StringField is not lossless: guessed charset, U+FFFD substitution, and a character-count length

pcapkit/corekit/fields/strings.py

@@ -12,6 +12,7 @@
 __all__ = [
     '_TextField',
     'StringField',
+    'DecodedString',
     'BitField',
     'PaddingField',
 ]
@@ -101,20 +102,34 @@ class StringField(_TextField[str]):
         length: Field size (in bytes); if a callable is given, it should return
             an integer value and accept the current packet as its only argument.
         default: Field default value, if any.
-        encoding: The encoding with which to decode the :obj:`bytes`.
-            If not provided, :mod:`pcapkit` will first try detecting its encoding
-            using |chardet|_. The fallback encoding is **UTF-8**.
+        encoding: The encoding with which to decode the :obj:`bytes`, and with which
+            a :obj:`str` is encoded when packing. If not provided, :mod:`pcapkit`
+            will first try detecting its encoding using |chardet|_. The fallback
+            encoding is **UTF-8**.
         errors: The error handling scheme to use for the handling of decoding errors.
-            The default is ``'strict'`` meaning that decoding errors raise a
-            :exc:`UnicodeDecodeError`. Other possible values are ``'ignore'`` and ``'replace'``
-            as well as any other name registered with :func:`codecs.register_error` that
-            can handle :exc:`UnicodeDecodeError`.
+            The default is ``'strict'``. Should decoding fail under it, the field
+            retries with ``'replace'`` rather than raising, so an octet the charset
+            cannot decode reads as U+FFFD -- the octets themselves are not lost,
+            see below. A codec that supports no error handler at all still raises:
+            ``idna`` reports ``Unsupported error handling: replace``, measured.
+            Other possible values are ``'ignore'`` and ``'replace'`` as well as any
+            other name registered with :func:`codecs.register_error` that can handle
+            :exc:`UnicodeDecodeError`.
         unquote: Whether to unquote the decoded string as a URL. Should decoding fail,
             the method will try again replacing ``'%'`` with ``'\x'`` then decoding the
             ``url`` as ``'utf-8'`` with ``'replace'`` for error handling.
         callback: Callback function to be called upon
             :meth:`self.__call__ <pcapkit.corekit.fields.field.FieldBase.__call__>`.
 
+    Unpacking is lossless. Where re-encoding the decoded :obj:`str` would not
+    reproduce the octets it came from -- a detected charset other than UTF-8, an
+    octet the charset cannot decode, a lossy ``errors`` scheme, or a URL quoting
+    that differs from the original -- the field returns a :class:`DecodedString`,
+    which remembers those octets and packs back as exactly them. The decoded
+    *text* is whatever it has always been, U+FFFD and all: a consumer that only
+    reads the text, such as a dumper writing it back out as UTF-8, sees no
+    change. A length of ``-1`` is resolved to the number of *encoded* octets.
+
     .. |chardet| replace:: ``chardet``
     .. _chardet: https://chardet.readthedocs.io
 
@@ -131,6 +146,20 @@ def __init__(self, length: 'int | Callable[[dict[str, Any]], int]',
         self._errors = errors
         self._unquote = unquote
 
+    def _encode(self, value: 'str') -> 'bytes':
+        """Encode a :obj:`str` value the way :meth:`pre_process` packs it.
+
+        Arguments:
+            value: Field value.
+
+        Returns:
+            Encoded field value.
+
+        """
+        if self._unquote:
+            value = urllib_parse.quote(value, encoding=self._encoding or 'utf-8', errors=self._errors)
+        return value.encode(self._encoding or 'utf-8', self._errors)
+
     def pre_process(self, value: 'str', packet: 'dict[str, Any]') -> 'bytes':  # pylint: disable=unused-argument
         """Process field value before construction (packing).
 
@@ -142,13 +171,18 @@ def pre_process(self, value: 'str', packet: 'dict[str, Any]') -> 'bytes':  # pyl
             Processed field value.
 
         """
-        if self._unquote:
-            value = urllib_parse.quote(value, encoding=self._encoding or 'utf-8', errors=self._errors)
-
+        # NOTE: The octets a DecodedString was decoded from are used verbatim,
+        # since re-encoding its text is exactly what would not reproduce them.
+        buffer = value.raw if isinstance(value, DecodedString) else self._encode(value)
+
+        # NOTE: A negative length is resolved from the *encoded* octets, not from
+        # the character count -- the ``Ns`` template counts octets, so counting
+        # characters truncated any text that does not encode one octet per
+        # character.
         if self._length < 0:
-            self._length = len(value)
+            self._length = len(buffer)
             self._template = f'{self._length}s'
-        return value.encode(self._encoding or 'utf-8', self._errors)
+        return buffer
 
     def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'str':  # pylint: disable=unused-argument
         """Process field value after parsing (unpacked).
@@ -158,7 +192,8 @@ def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'str':  # py
             packet: Packet data.
 
         Returns:
-            Processed field value.
+            Processed field value; a :class:`DecodedString` carrying ``value``
+            where :meth:`pre_process` would not otherwise reproduce it.
 
         """
         if self._unquote:
@@ -172,7 +207,51 @@ def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'str':  # py
                 ret = value.decode(charset, self._errors)
             except UnicodeError:
                 ret = value.decode(charset, 'replace')
-        return ret
+
+        # NOTE: Encoding may itself fail -- ``errors='strict'`` over text that
+        # decoded under a lone-surrogate scheme, for one -- which is as much a
+        # failure to reproduce the octets as a mismatch is.
+        try:
+            lossless = self._encode(ret) == value
+        except UnicodeError:
+            lossless = False
+        if lossless:
+            return ret
+        return DecodedString(ret, value)
+
+
+class DecodedString(str):
+    """Decoded :class:`StringField` value that remembers its octets.
+
+    :class:`StringField` returns one where encoding the decoded text would not
+    reproduce the octets it was decoded from, so that the field packs it back as
+    exactly those octets. It is otherwise an ordinary :obj:`str`, and deliberately
+    so: it compares and hashes as its text, since the text is the value and
+    :attr:`raw` is provenance. Two of them with the same text and different octets
+    therefore compare equal -- the alternative, mixing :attr:`raw` into
+    :meth:`~object.__eq__`, would make ``info.comment == 'café'`` false against a
+    plain :obj:`str` literal, which is the comparison callers actually write.
+
+    Any string operation on one returns a plain :obj:`str`, which a field then
+    packs with its own encoding; that is intended, as an edited value is no longer
+    the text those octets held.
+
+    Args:
+        value: Decoded text.
+        raw: Octets ``value`` was decoded from.
+
+    """
+
+    #: Octets the string was decoded from.
+    raw: 'bytes'
+
+    def __new__(cls, value: 'str', raw: 'bytes') -> 'Self':
+        self = super().__new__(cls, value)
+        self.raw = raw
+        return self
+
+    def __getnewargs__(self) -> 'tuple[str, bytes]':  # type: ignore[override]
+        return str(self), self.raw
 
 
 class BitField(_TextField[Dict[str, Any]]):

tests/corekit/test_stringfield_lossless_unit.py

@@ -0,0 +1,195 @@
+# -*- coding: utf-8 -*-
+"""StringField packs back exactly the octets it unpacked.
+
+GitHub issue #1293: :class:`~pcapkit.corekit.fields.strings.StringField` was
+lossy three ways. Without an ``encoding`` it decoded with the charset
+:func:`~pcapkit.utilities.chardet.detect` guessed and packed the text back as
+UTF-8; an octet the charset could not decode became U+FFFD and was packed as
+U+FFFD's own three octets; and a length of ``-1`` was taken from the *character*
+count, so multi-octet text lost its tail.
+
+The octets are now remembered by a :class:`DecodedString
+<pcapkit.corekit.fields.strings.DecodedString>` and packed verbatim, while the
+decoded text is left exactly as it was -- U+FFFD and all. That the text is
+unchanged is load-bearing rather than incidental, so
+:class:`TestStringFieldDumpers` runs the real ``plist`` and ``tree`` dumpers over
+an undecodable octet: a lone surrogate there instead of U+FFFD raises
+``UnicodeEncodeError`` inside :mod:`dictdumper`, which is a crash on arbitrary
+capture bytes.
+
+:mod:`pcapkit` is imported inside each test, after
+:func:`~tests._support.reimport_once_per_class`, not at module load.
+
+"""
+
+import copy
+import os
+import pickle
+import tempfile
+import unittest
+
+from tests._support import reimport_once_per_class
+
+#: Charsets whose codecs reject a ``'surrogateescape'`` error handler, so that a
+#: fallback through one would raise where ``'replace'`` does not.
+NO_SURROGATE_CHARSETS = ('utf-16', 'utf-16-le', 'utf-16-be', 'utf-32', 'utf-7')
+
+
+def pcapng(comment: bytes) -> bytes:
+    """A minimal pcapng: SHB, IDB and one EPB, the first two carrying ``comment``.
+
+    Args:
+        comment: Octets to write as the ``opt_comment`` of the SHB and the IDB.
+
+    Returns:
+        The capture file's octets.
+
+    """
+    opts = (b'\x01\x00' + len(comment).to_bytes(2, 'little')
+            + comment + bytes((-len(comment)) % 4) + b'\x00\x00\x00\x00')
+
+    shb_body = bytes.fromhex('4d3c2b1a') + b'\x01\x00\x00\x00' + b'\xff' * 8 + opts
+    shb_len = (12 + len(shb_body)).to_bytes(4, 'little')
+    shb = bytes.fromhex('0a0d0d0a') + shb_len + shb_body + shb_len
+
+    idb_body = b'\x01\x00\x00\x00' + (0x40000).to_bytes(4, 'little') + opts
+    idb_len = (12 + len(idb_body)).to_bytes(4, 'little')
+    idb = b'\x01\x00\x00\x00' + idb_len + idb_body + idb_len
+
+    packet = bytes.fromhex('ffffffffffff' '000000000001' '0800') + bytes(34)
+    epb_body = (bytes(12) + len(packet).to_bytes(4, 'little')
+                + len(packet).to_bytes(4, 'little') + packet)
+    epb_len = (12 + len(epb_body)).to_bytes(4, 'little')
+    epb = b'\x06\x00\x00\x00' + epb_len + epb_body + epb_len
+
+    return shb + idb + epb
+
+
+class TestStringFieldLossless(unittest.TestCase):
+    """Pin ``pack(unpack(octets)) == octets`` for :class:`StringField`."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def roundtrip(self, octets: bytes, **kwargs: object) -> 'tuple[str, bytes]':
+        """Unpack ``octets`` through a field of their own width, then pack it back."""
+        from pcapkit.corekit.fields.strings import StringField
+
+        field = StringField(length=len(octets), **kwargs)  # type: ignore[arg-type]
+        value = field.unpack(octets, {})
+        return value, field.pack(value, {})
+
+    def test_detected_charset_is_reused_on_pack(self) -> None:
+        value, packed = self.roundtrip(bytes.fromhex('e974e9'))
+        self.assertEqual(packed.hex(), 'e974e9')
+        self.assertEqual(value, 'été')
+
+    def test_undecodable_octet_round_trips_as_replacement_text(self) -> None:
+        value, packed = self.roundtrip(b'caf\xe9', encoding='utf-8')
+        self.assertEqual(packed, b'caf\xe9')
+        self.assertEqual(value, 'caf�')  # the text main decoded, unchanged
+
+    def test_lossy_errors_scheme_still_round_trips(self) -> None:
+        for errors in ('replace', 'ignore'):
+            with self.subTest(errors=errors):
+                _, packed = self.roundtrip(b'a\xffb', encoding='utf-8', errors=errors)
+                self.assertEqual(packed, b'a\xffb')
+
+    def test_decoded_text_never_holds_a_lone_surrogate(self) -> None:
+        for octets in (b'\xff\xfe', b'caf\xe9', b'\x80', bytes(range(128, 160)),
+                       bytes.fromhex('e974e9'), b'\xed\xa0\x80'):
+            for kwargs in ({}, {'encoding': 'utf-8'}, {'unquote': True}):
+                with self.subTest(octets=octets, kwargs=tuple(sorted(kwargs))):
+                    value, packed = self.roundtrip(octets, **kwargs)  # type: ignore[arg-type]
+                    self.assertEqual(packed, octets)
+                    self.assertFalse([char for char in value if 0xD800 <= ord(char) <= 0xDFFF])
+
+    def test_charsets_without_surrogateescape_do_not_raise(self) -> None:
+        for charset in NO_SURROGATE_CHARSETS:
+            with self.subTest(charset=charset):
+                _, packed = self.roundtrip(b'\xff\xfe\x41', encoding=charset)
+                self.assertEqual(packed, b'\xff\xfe\x41')
+
+    def test_valid_text_decodes_to_a_plain_str(self) -> None:
+        value, packed = self.roundtrip('héllo'.encode())
+        self.assertIs(type(value), str)
+        self.assertEqual(value, 'héllo')
+        self.assertEqual(packed, 'héllo'.encode())
+
+    def test_unquote_keeps_the_original_quoting(self) -> None:
+        value, packed = self.roundtrip(b'a%2fb%2F', unquote=True)
+        self.assertEqual(value, 'a/b/')
+        self.assertEqual(packed, b'a%2fb%2F')
+
+    def test_variable_length_counts_encoded_octets(self) -> None:
+        from pcapkit.corekit.fields.strings import StringField
+
+        self.assertEqual(StringField(length=-1).pack('héllo', {}), 'héllo'.encode())
+
+    def test_edited_value_packs_with_the_field_encoding(self) -> None:
+        from pcapkit.corekit.fields.strings import StringField
+
+        field = StringField(length=-1)
+        value = StringField(length=3).unpack(bytes.fromhex('e974e9'), {})
+        self.assertEqual(field.pack(value.upper(), {}), 'ÉTÉ'.encode())
+
+    def test_decoded_string_survives_copy_and_pickle(self) -> None:
+        from pcapkit.corekit.fields.strings import StringField
+
+        field = StringField(length=3)
+        value = field.unpack(bytes.fromhex('e974e9'), {})
+        for clone in (copy.copy(value), copy.deepcopy(value), pickle.loads(pickle.dumps(value))):
+            with self.subTest(clone=type(clone).__name__):
+                self.assertEqual(clone, 'été')
+                self.assertEqual(field.pack(clone, {}).hex(), 'e974e9')
+
+    def test_decoded_string_compares_and_hashes_as_its_text(self) -> None:
+        from pcapkit.corekit.fields.strings import DecodedString
+
+        value = DecodedString('été', b'\xe9t\xe9')
+        self.assertEqual(value, 'été')
+        self.assertEqual(hash(value), hash('été'))
+        self.assertEqual({value: 1}['été'], 1)
+
+
+class TestStringFieldDumpers(unittest.TestCase):
+    """The dumpers render an undecodable octet rather than failing on it."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_plist_and_tree_dump_an_undecodable_comment(self) -> None:
+        import pcapkit
+
+        for comment in (b'\xff\xfe', b'caf\xe9'):
+            for fmt, suffix in (('plist', '.plist'), ('tree', '.txt'), ('json', '.json')):
+                with self.subTest(comment=comment, format=fmt):
+                    with tempfile.TemporaryDirectory() as tmp:
+                        fin = os.path.join(tmp, 'in.pcapng')
+                        fout = os.path.join(tmp, 'out' + suffix)
+                        with open(fin, 'wb') as file:
+                            file.write(pcapng(comment))
+
+                        pcapkit.extract(fin=fin, fout=fout, format=fmt,
+                                        store=False, engine='default')
+                        with open(fout, 'rb') as file:
+                            dumped = file.read()
+                    # The dump is readable UTF-8 -- a lone surrogate would have
+                    # raised in the writer rather than reaching the file -- and the
+                    # comment is in it, as U+FFFD or as the escape ``json`` writes.
+                    text = dumped.decode('utf-8')
+                    self.assertTrue('�' in text or '\\ufffd' in text)
+
+    def test_the_dumped_comment_still_packs_back_to_its_octets(self) -> None:
+        from pcapkit.protocols.misc.pcapng import PCAPNG
+
+        whole = pcapng(b'caf\xe9')
+        shb = whole[:int.from_bytes(whole[4:8], 'little')]
+
+        parsed = PCAPNG(shb, len(shb), num=1, sct=1, ctx=None)
+        self.assertEqual(parsed.info.options[1].comment, 'caf�')
+        self.assertEqual(PCAPNG.from_data(parsed.info, num=1, sct=1, ctx=None).data, shb)
+
+
+if __name__ == '__main__':
+    unittest.main()