diff --git a/.secrets.baseline b/.secrets.baseline index 5ac20045..ddd47d12 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -222,7 +222,7 @@ "filename": "src/cachekit/cache_handler.py", "hashed_secret": "5baa61e4c9b93f3f0682250b6cf8331b7ee68fd8", "is_verified": false, - "line_number": 452 + "line_number": 455 } ], "src/cachekit/config/decorator.py": [ @@ -887,5 +887,5 @@ } ] }, - "generated_at": "2026-09-06T23:40:47Z" + "generated_at": "2026-09-13T12:55:39Z" } diff --git a/src/cachekit/cache_handler.py b/src/cachekit/cache_handler.py index 590d2d9e..0f47dfcd 100644 --- a/src/cachekit/cache_handler.py +++ b/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: diff --git a/src/cachekit/decorators/wrapper.py b/src/cachekit/decorators/wrapper.py index 73bcc682..08b2209b 100644 --- a/src/cachekit/decorators/wrapper.py +++ b/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) diff --git a/src/cachekit/serializers/auto_serializer.py b/src/cachekit/serializers/auto_serializer.py index f75f7bb0..404a5d41 100644 --- a/src/cachekit/serializers/auto_serializer.py +++ b/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 diff --git a/src/cachekit/serializers/base.py b/src/cachekit/serializers/base.py index f076b0a0..5c4feba4 100644 --- a/src/cachekit/serializers/base.py +++ b/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. diff --git a/tests/unit/test_auto_serializer_mutation_and_corruption.py b/tests/unit/test_auto_serializer_mutation_and_corruption.py index bfb41f30..061d89c0 100644 --- a/tests/unit/test_auto_serializer_mutation_and_corruption.py +++ b/tests/unit/test_auto_serializer_mutation_and_corruption.py @@ -21,13 +21,17 @@ from __future__ import annotations import functools +import logging +from unittest import mock import msgpack import pytest from cachekit._rust_serializer import ByteStorage +from cachekit.cache_handler import CacheOperationHandler, CacheSerializationHandler, handle_decrypt_failure +from cachekit.key_generator import CacheKeyGenerator from cachekit.serializers import AutoSerializer -from cachekit.serializers.base import SerializationError +from cachekit.serializers.base import ERROR_ECHO_MAX, SerializationError, bounded_error # Requires the [data] extra — absent e.g. in the free-threaded CI lane until # numpy/pandas ship free-threaded wheels (LAB-511). @@ -208,6 +212,18 @@ def test_ndarray_where_the_writer_emits_a_list_or_dict_is_refused(self, kind: st with pytest.raises(SerializationError, match="Forged columnar payload"): AutoSerializer().deserialize(_entry(kind, body)) + @pytest.mark.parametrize( + "kind, body", + [("dataframe", [1, 2]), ("series", 7)], + ids=["dataframe-body-is-list", "series-body-is-int"], + ) + def test_non_dict_document_is_refused(self, kind: str, body: object) -> None: + # The writer always emits a dict body; a forged non-dict decodes cleanly under the msgpack + # bound and now hits the _expect shape gate directly (the dead bytes-preamble that used to + # sit ahead of it is gone), so the "document is " guard is reachable in production. + with pytest.raises(SerializationError, match="Forged columnar payload: document is"): + AutoSerializer().deserialize(_entry(kind, body)) # type: ignore[arg-type] + def _numpy_raw(dtype: bytes, shape: bytes) -> bytes: return b"NUMPY_RAW" + len(dtype).to_bytes(2, "little") + dtype + len(shape).to_bytes(2, "little") + shape @@ -240,3 +256,74 @@ def test_well_formed_numpy_still_round_trips(self) -> None: arr = np.arange(6, dtype=" SerializationError: + """The SerializationError raised by decoding a poisoned columnar entry that carries a 1 MiB + column name and a 4 KB forged dtype — the real error object the read-path log sites echo.""" + big_name = "n" * (1024 * 1024) # 1 MiB column name (already capped in the field echo, #276) + big_dtype = "z" * 4096 # 4 KB forged dtype — numpy echoes the whole string, uncapped (LAB-3131) + body = { + "columns": [big_name], + "index": None, + "data": {big_name: {"type": "numeric", "data": b"", "dtype": big_dtype}}, + } + with pytest.raises(SerializationError) as excinfo: + AutoSerializer().deserialize(_entry("dataframe", body)) + return excinfo.value + + +@pytest.mark.unit +class TestForgedEntryErrorEchoIsBounded: + """LAB-3131 AC1: a poisoned columnar entry of any size logs O(1)-bounded text at every wrap + site. #276 capped the per-field marker/column echoes, but ``_dtype_from_untrusted`` still + echoed the full forged dtype, so the SerializationError message — and every log line built + from it — grew with the payload. The bound is applied once, at each read-path wrap site, via + :func:`bounded_error`. + """ + + def test_bounded_error_clips_and_neutralizes_control_chars(self) -> None: + # Over-length text is clipped to O(1) with the true length preserved for forensics. + clipped = bounded_error(SerializationError("x" * (1024 * 1024))) + assert len(clipped) <= ERROR_ECHO_MAX + 64 + assert "1048576 chars total" in clipped + # Every line/terminal-control char is escaped, so the result is one terminal-safe line — + # not just \n/\r (ANSI \x1b, vertical tab \x0b, Unicode line-sep U+2028 all handled). + raw = "a\nb\rc\x1bd\x0be" + chr(0x2028) + "f" # newline, CR, ANSI ESC, VT, U+2028 line-sep + unsafe = bounded_error(SerializationError(raw)) + assert not any(ch in unsafe for ch in "\n\r\x1b\x0b" + chr(0x2028)) + assert "\\x1b" in unsafe + + def test_forged_dtype_produces_an_unbounded_message(self) -> None: + # Guards the premise: without the bound the echoed text really is huge (the 4 KB dtype is + # in there in full), so the assertions below are proving the bound does real work. + assert len(str(_oversized_forged_error())) > 4096 + + def test_handle_decrypt_failure_warning_line_is_bounded(self, caplog) -> None: + # _handle_l2_read_error and the wrapper L1 SerializationError guard both route the poisoned + # error here; this is the WARNING line emitted on every poisoned read. + err = _oversized_forged_error() + with caplog.at_level(logging.WARNING): + handle_decrypt_failure(err, tier="l2", cache_key="ns:app:key", fail_closed=False) + lines = [r.getMessage() for r in caplog.records if "decrypt/integrity failure" in r.getMessage()] + # Line = fixed template + one bounded_error() echo, so it is O(1) in ERROR_ECHO_MAX, + # independent of the (multi-MB) forged payload. + assert lines and all(len(line) < ERROR_ECHO_MAX + 256 for line in lines) + + def test_end_to_end_l2_read_of_forged_entry_logs_bounded_line(self, caplog) -> None: + # The real read plumbing: get_cached_value -> _handle_l2_read_error -> handle_decrypt_failure. + # Catches a regression if a future edit logs the poisoned error ahead of the bounded site. + err = _oversized_forged_error() + serialization = mock.MagicMock(spec=CacheSerializationHandler) + serialization.deserialize_data.side_effect = err + serialization.encryption_fail_closed = False # real bool: a MagicMock is truthy -> fail-closed + serialization.supports_mmap_read.return_value = False + handler = CacheOperationHandler(serialization, CacheKeyGenerator()) + backend = mock.MagicMock() + backend.get.return_value = b"poisoned-entry-bytes" + handler.set_cache_handler(backend) + + with caplog.at_level(logging.WARNING): + assert handler.get_cached_value("ns:app:key") is None # fail-open miss + lines = [r.getMessage() for r in caplog.records if "decrypt/integrity failure" in r.getMessage()] + assert lines and all(len(line) < ERROR_ECHO_MAX + 256 for line in lines) diff --git a/tests/unit/test_auto_serializer_new_types.py b/tests/unit/test_auto_serializer_new_types.py index 7009f920..e4dd25fa 100644 --- a/tests/unit/test_auto_serializer_new_types.py +++ b/tests/unit/test_auto_serializer_new_types.py @@ -729,7 +729,9 @@ def test_nullable_dtypes_and_objects_roundtrip(self): ) data = ser._serialize_dataframe(df) # previously raised: msgpack can't pack pd.NA - out = ser._deserialize_dataframe(data) + # _decode_columnar decodes the msgpack body then hands the document to + # _deserialize_dataframe (which now takes a decoded document, not bytes). + out = ser._decode_columnar(data, "dataframe") assert list(out.columns) == ["ints", "floats", "objs", "plain"] assert out.shape == (4, 4) @@ -749,7 +751,7 @@ def test_pyarrow_backed_dtype_does_not_crash(self): df = pd.DataFrame({"x": pd.array([1, 2, 3], dtype="int64[pyarrow]")}) data = ser._serialize_dataframe(df) - out = ser._deserialize_dataframe(data) + out = ser._decode_columnar(data, "dataframe") assert out["x"].tolist() == [1, 2, 3] @@ -759,7 +761,7 @@ def test_nullable_series_roundtrip(self): s = pd.Series(pd.array([1, 2, None, 4], dtype="Int64"), name="n") data = ser._serialize_series(s) # previously raised on the pd.NA sentinel - out = ser._deserialize_series(data) + out = ser._decode_columnar(data, "series") assert out.name == "n" assert out.iloc[0] == 1 and out.iloc[3] == 4 @@ -771,6 +773,6 @@ def test_pyarrow_backed_series_does_not_crash(self): ser = AutoSerializer(enable_integrity_checking=False) s = pd.Series(pd.array([1, 2, 3], dtype="int64[pyarrow]"), name="x") - out = ser._deserialize_series(ser._serialize_series(s)) + out = ser._decode_columnar(ser._serialize_series(s), "series") assert out.tolist() == [1, 2, 3]