fix(serializers): bound forged-entry error echoes; retire columnar dead code (LAB-3131)

src/cachekit/cache_handler.py

@@ -36,6 +36,7 @@
     SerializationFormat,
     SerializationMetadata,
     SuspiciousCacheEntryError,
+    bounded_error,
 )
 from cachekit.serializers.encryption_wrapper import (
     DecryptionAuthenticationError,
@@ -173,10 +174,12 @@ def handle_decrypt_failure(error: Exception, *, tier: str, cache_key: str, fail_
     if fail_closed and isinstance(error, DecryptionAuthenticationError):
         get_logger().error(
             f"{tier.upper()} cache decrypt AUTHENTICATION failure for {redact_cache_key(cache_key)}; "
-            f"failing closed (encryption.fail_closed=True): {error}"
+            f"failing closed (encryption.fail_closed=True): {bounded_error(error)}"
         )
         raise error
-    get_logger().warning(f"{tier.upper()} cache decrypt/integrity failure ({reason}) for {redact_cache_key(cache_key)}: {error}")
+    get_logger().warning(
+        f"{tier.upper()} cache decrypt/integrity failure ({reason}) for {redact_cache_key(cache_key)}: {bounded_error(error)}"
+    )
     return reason
 
 
@@ -1166,8 +1169,8 @@ def deserialize_data(self, data: str | bytes | memoryview, cache_key: str = "")
             # SerializationError/EncryptionError: let the outer handler log and handle
             raise
         except Exception as e:
-            get_logger().error(f"Deserialization failed with {self.serializer_name}: {e}")
-            raise SerializationError(f"Failed to deserialize data with {self.serializer_name}: {e}") from e
+            get_logger().error(f"Deserialization failed with {self.serializer_name}: {bounded_error(e)}")
+            raise SerializationError(f"Failed to deserialize data with {self.serializer_name}: {bounded_error(e)}") from e
 
     def _deserialize_interop(self, data: str | bytes | memoryview, cache_key: str) -> Any:
         """Interop/v1 read path: config decides encryption, never the stored bytes.
@@ -1211,8 +1214,8 @@ def _deserialize_interop(self, data: str | bytes | memoryview, cache_key: str) -
         except (ValueError, SerializationError):
             raise
         except Exception as e:
-            get_logger().error(f"Interop deserialization failed: {e}")
-            raise SerializationError(f"Failed to deserialize interop cache entry: {e}") from e
+            get_logger().error(f"Interop deserialization failed: {bounded_error(e)}")
+            raise SerializationError(f"Failed to deserialize interop cache entry: {bounded_error(e)}") from e
 
 
 class CacheOperationHandler:

src/cachekit/decorators/wrapper.py

@@ -36,7 +36,7 @@
 from ..l1_cache import DEFAULT_L1_TTL_SECONDS, get_l1_cache
 from ..object_cache import ObjectCache
 from ..reliability import CircuitBreakerConfig
-from ..serializers.base import SerializationError
+from ..serializers.base import SerializationError, bounded_error
 from ..serializers.encryption_wrapper import DecryptionAuthenticationError, KeyringConfigurationError
 
 # Config import removed - using direct DecoratorConfig integration
@@ -1292,7 +1292,7 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:  # noqa: PLR0912
                     raise
                 except Exception as e:
                     # L1 deserialization failed - invalidate and continue to L2
-                    logger().warning(f"L1 cache deserialization failed for {cache_key}: {e}")
+                    logger().warning(f"L1 cache deserialization failed for {cache_key}: {bounded_error(e)}")
                     _l1_cache.invalidate(cache_key)
 
         # Continue with the rest of the sync wrapper logic...
@@ -1648,7 +1648,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                         raise
                     except Exception as e:
                         # L1 deserialization failed - invalidate and continue to L2
-                        logger().warning(f"L1 cache deserialization failed for {cache_key}: {e}")
+                        logger().warning(f"L1 cache deserialization failed for {cache_key}: {bounded_error(e)}")
                         _l1_cache.invalidate(cache_key)
 
             # Initialize backend only when needed (lazy init for performance)

src/cachekit/serializers/auto_serializer.py

@@ -202,6 +202,23 @@ def _column_values(info: dict[str, Any], what: str) -> Any:
     raise SerializationError(f"Forged columnar payload: {what} type is {shown!r:.40}, expected 'numeric' or 'object'")
 
 
+def _column_trio(series: Any) -> dict[str, Any]:
+    """Build one column's/Series' marker — the write-side mirror of :func:`_column_values`.
+
+    The marker is the 3-key ``{type: "numeric", data, dtype}`` for a plain-numeric column and the
+    2-key ``{type: "object", data}`` otherwise (the ``dtype`` key is numeric-only) — "trio" names
+    the maximal numeric form. One writer for both the DataFrame-column and the bare-Series paths, so
+    the marker set and key order live in a single place and a third marker cannot be added to one
+    side only (the AC3 goal). Plain NumPy
+    numeric dtypes take the raw-buffer path; everything else (nullable/extension dtypes) takes the
+    NA-safe object path so pd.NA/NaT do not crash msgpack (#160). Wire bytes and key order MUST
+    stay byte-identical to the interop fixtures.
+    """
+    if _is_plain_numpy_numeric(series.dtype):
+        return {"type": "numeric", "data": series.values.tobytes(), "dtype": str(series.dtype)}  # type: ignore[union-attr]
+    return {"type": "object", "data": _na_safe_object_list(series)}
+
+
 def _na_safe_object_list(series: Any) -> list:
     """``series.tolist()`` with scalar pandas NA sentinels (pd.NA/NaT/NaN) mapped to None.
 
@@ -804,15 +821,10 @@ def _serialize_dataframe(self, df: pd.DataFrame) -> bytes:
             "data": {},
         }
 
-        # Serialize each column separately
+        # Serialize each column separately — _column_trio is the single writer shared with
+        # _serialize_series and mirrored by the _column_values decoder.
         for col in df.columns:
-            series = df[col]
-            # Fast raw-buffer path only for plain NumPy numeric dtypes; nullable/extension
-            # dtypes fall through to the NA-safe object path (see helper docstrings).
-            if _is_plain_numpy_numeric(series.dtype):
-                serialized["data"][col] = {"type": "numeric", "data": series.values.tobytes(), "dtype": str(series.dtype)}  # type: ignore[union-attr]
-            else:
-                serialized["data"][col] = {"type": "object", "data": _na_safe_object_list(series)}
+            serialized["data"][col] = _column_trio(df[col])
 
         msgpack_data = msgpack.packb(serialized, **self._msgpack_pack_opts)
 
@@ -821,8 +833,13 @@ def _serialize_dataframe(self, df: pd.DataFrame) -> bytes:
         else:
             return msgpack_data  # type: ignore[return-value]
 
-    def _deserialize_dataframe(self, data) -> pd.DataFrame:
-        """Deserialize DataFrame from column-wise data.
+    def _deserialize_dataframe(self, document) -> pd.DataFrame:
+        """Rebuild a DataFrame from the already-decoded columnar ``document``.
+
+        ``document`` is the msgpack-decoded body — every production caller (the verified-envelope
+        route in :meth:`deserialize` and :meth:`_decode_columnar`) decodes under
+        ``unpackb_bounded`` first, so this method never touches the wire bytes and never re-runs
+        the decode bound. A forged non-dict body is refused by the ``_expect`` shape gate.
 
         Requires: pandas installed (HAS_PANDAS=True)
 
@@ -833,14 +850,7 @@ def _deserialize_dataframe(self, data) -> pd.DataFrame:
         if not HAS_PANDAS:
             raise RuntimeError("Pandas not installed. Install with: pip install cachekit[data]")
 
-        # If data is already unpacked (from Rust layer), use it directly
-        if isinstance(data, dict):
-            serialized = data
-        else:
-            # Otherwise unpack msgpack
-            serialized = unpackb_bounded(data, **self._msgpack_unpack_opts)
-
-        serialized = _expect(serialized, dict, "document")
+        serialized = _expect(document, dict, "document")
         columns_data = {}
         for col, col_info in _expect(serialized["data"], dict, "data").items():
             what = f"column {col!r:.40}"  # col is attacker-chosen: cap the echo
@@ -868,13 +878,9 @@ def _serialize_series(self, series: pd.Series) -> bytes:
             "index": series.index.tolist() if series.index.name or not series.index.equals(pd.RangeIndex(len(series))) else None,
         }
 
-        # Same dtype handling as _serialize_dataframe: plain NumPy numeric uses the raw
-        # buffer; nullable/extension dtypes take the NA-safe object path so pd.NA/NaT
-        # do not crash msgpack (#160).
-        if _is_plain_numpy_numeric(series.dtype):
-            serialized.update({"type": "numeric", "data": series.values.tobytes(), "dtype": str(series.dtype)})  # type: ignore[union-attr]
-        else:
-            serialized.update({"type": "object", "data": _na_safe_object_list(series)})
+        # Same {type, data[, dtype]} trio as each DataFrame column, appended after name/index
+        # so the on-wire key order is {name, index, type, data[, dtype]} (byte-compatible).
+        serialized.update(_column_trio(series))
 
         msgpack_data = msgpack.packb(serialized, **self._msgpack_pack_opts)
 
@@ -883,8 +889,12 @@ def _serialize_series(self, series: pd.Series) -> bytes:
         else:
             return msgpack_data  # type: ignore[return-value]
 
-    def _deserialize_series(self, data) -> pd.Series:
-        """Deserialize Pandas Series.
+    def _deserialize_series(self, document) -> pd.Series:
+        """Rebuild a Series from the already-decoded columnar ``document``.
+
+        ``document`` is the msgpack-decoded body; the same decoded-only contract as
+        :meth:`_deserialize_dataframe` (its callers run ``unpackb_bounded`` first). A forged
+        non-dict body is refused by the ``_expect`` shape gate.
 
         Requires: pandas installed (HAS_PANDAS=True)
 
@@ -895,14 +905,7 @@ def _deserialize_series(self, data) -> pd.Series:
         if not HAS_PANDAS:
             raise RuntimeError("Pandas not installed. Install with: pip install cachekit[data]")
 
-        # If data is already unpacked (from Rust layer), use it directly
-        if isinstance(data, dict):
-            serialized = data
-        else:
-            # Otherwise unpack msgpack
-            serialized = unpackb_bounded(data, **self._msgpack_unpack_opts)
-
-        serialized = _expect(serialized, dict, "document")
+        serialized = _expect(document, dict, "document")
         series = pd.Series(_column_values(serialized, "series"), name=serialized["name"])
 
         # Restore index if it was serialized

src/cachekit/serializers/base.py

@@ -349,6 +349,42 @@ class SuspiciousCacheEntryError(SerializationError):
 #: optional dependency — is an environment fault, not a bad cache entry, and must bubble.
 PAYLOAD_DECODE_ERRORS = (ValueError, TypeError, KeyError, AttributeError, OverflowError, BufferError, SyntaxError)
 
+#: Character cap for any untrusted-payload-derived error text that a read path logs or folds into
+#: an outer message. A forged/corrupt cache entry controls the exception text — numpy echoes a
+#: whole forged dtype string, an inner ``{e}`` wrap carries it up — so it is unbounded. Measured
+#: on the pre-#276 code, an 8.3 KB envelope carrying a 1 MiB column name produced a 4.19 MB log
+#: line (505x); post-#276 the marker/column echoes are capped but ``_dtype_from_untrusted`` still
+#: echoes the full dtype (a 1 MiB forged dtype still floods). ``repr`` escapes newlines but the
+#: read paths log ``str(e)``, so :func:`bounded_error` also collapses them — one poisoned read is
+#: always exactly one bounded log line (LAB-3131).
+ERROR_ECHO_MAX = 512
+
+#: Every char that could split a log record into extra lines or spoof a terminal, escaped so an
+#: attacker-controlled error message is always ONE terminal-safe line: all C0 controls (incl.
+#: ``\n``/``\r``/``\x0b``/``\x0c``), DEL, the C1 range (incl. NEL ``\x85``), and the Unicode
+#: line/paragraph separators ``U+2028``/``U+2029``. Applied by :func:`bounded_error` after the
+#: length clip, so the escape expansion works on a bounded string, never the raw payload.
+_LOG_UNSAFE_ESCAPES = {c: f"\\x{c:02x}" for c in (*range(0x20), 0x7F, *range(0x80, 0xA0))}
+_LOG_UNSAFE_ESCAPES.update({0x2028: "\\u2028", 0x2029: "\\u2029"})
+
+
+def bounded_error(exc: BaseException) -> str:
+    """``str(exc)`` clipped to :data:`ERROR_ECHO_MAX` and reduced to one terminal-safe line, for
+    logging or re-wrapping a failure whose text is influenced by untrusted cache bytes.
+
+    Applied once at each trust-boundary wrap site (the read-path log/re-raise points in
+    ``cache_handler``/``decorators.wrapper``) rather than per field: the bound then holds for
+    every attacker-inflatable field — marker, column name, dtype — including ones a future field
+    would add. Over-length text is truncated with the true length appended so the log still says
+    "this was huge", then every line/terminal-control char is escaped (:data:`_LOG_UNSAFE_ESCAPES`)
+    so one poisoned read is always exactly one log line with no injected ANSI or newlines. Clipping
+    before escaping keeps output O(1) (escape expansion applies to at most ``ERROR_ECHO_MAX`` chars).
+    """
+    text = str(exc)
+    if len(text) > ERROR_ECHO_MAX:
+        text = f"{text[:ERROR_ECHO_MAX]}… [truncated, {len(text)} chars total]"
+    return text.translate(_LOG_UNSAFE_ESCAPES)
+
 
 def unpackb_bounded(data: bytes | bytearray | memoryview, **unpack_opts: Any) -> Any:
     """Decode one untrusted MessagePack document under cachekit-owned bounds.