fix: escape backslashes in SQL literals (poisoned-buffer on common payloads) + fallback hardening

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "model-ledger"
-version = "0.7.11"
+version = "0.7.12"
 description = "Developer-first model inventory and governance framework for SR 11-7, EU AI Act, and NIST AI RMF compliance"
 readme = "README.md"
 requires-python = ">=3.10"

src/model_ledger/backends/snowflake.py

@@ -6,7 +6,6 @@
 
 from __future__ import annotations
 
-import contextlib
 import json
 import logging
 import threading
@@ -67,7 +66,13 @@ def _exec_no_result(session: Any, sql: str) -> None:
 def _esc(value: str | None) -> str:
     if value is None:
         return "NULL"
-    return "'" + str(value).replace("'", "''") + "'"
+    # Backslashes FIRST: Snowflake processes backslash escape sequences inside
+    # single-quoted string constants, so a JSON payload containing \" (every
+    # embedded double quote, as serialized by json.dumps) or \\ reaches the
+    # server mangled — PARSE_JSON then fails or silently corrupts the value.
+    # Empirically verified 2026-07-10: PARSE_JSON('{"a": "x \" y"}') errors;
+    # the doubled form round-trips.
+    return "'" + str(value).replace("\\", "\\\\").replace("'", "''") + "'"
 
 
 def _is_privilege_error(exc: Exception) -> bool:
@@ -276,30 +281,31 @@ def _flush_models_pandas(self) -> bool:
         if not hasattr(conn, "_session_parameters"):
             return False
 
-        df = pd.DataFrame(
-            [
-                {
-                    "MODEL_HASH": m.model_hash,
-                    "NAME": m.name,
-                    "OWNER": m.owner,
-                    "MODEL_TYPE": m.model_type,
-                    "MODEL_ORIGIN": m.model_origin,
-                    "TIER": m.tier,
-                    "PURPOSE": m.purpose,
-                    "STATUS": m.status,
-                    "CREATED_AT": m.created_at.isoformat(),
-                    "LAST_SEEN": m.last_seen.isoformat() if m.last_seen else None,
-                    "METADATA": json.dumps(m.metadata, default=str) if m.metadata else None,
-                }
-                for m in self._model_buffer
-            ]
-        )
-
         staging = f"{self._schema}.MODELS_STAGING"
         # The bulk path needs CREATE TABLE on the schema for the staging table.
         # A least-privilege writer (INSERT/UPDATE/SELECT only) can't create it;
         # fall back to the DDL-free SQL MERGE path rather than failing the write.
+        # Row construction sits inside the try so a serialization surprise also
+        # falls back instead of poisoning the buffer.
         try:
+            df = pd.DataFrame(
+                [
+                    {
+                        "MODEL_HASH": m.model_hash,
+                        "NAME": m.name,
+                        "OWNER": m.owner,
+                        "MODEL_TYPE": m.model_type,
+                        "MODEL_ORIGIN": m.model_origin,
+                        "TIER": m.tier,
+                        "PURPOSE": m.purpose,
+                        "STATUS": m.status,
+                        "CREATED_AT": m.created_at.isoformat(),
+                        "LAST_SEEN": m.last_seen.isoformat() if m.last_seen else None,
+                        "METADATA": json.dumps(m.metadata, default=str) if m.metadata else None,
+                    }
+                    for m in self._model_buffer
+                ]
+            )
             self._exec_no_result(
                 f"CREATE OR REPLACE TEMPORARY TABLE {staging} LIKE {self._schema}.MODELS"
             )
@@ -322,21 +328,25 @@ def _flush_models_pandas(self) -> bool:
             self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
         except Exception as e:
             # The pandas path is an optimization, not a correctness
-            # requirement: privilege denials (no CREATE TEMPORARY TABLE) and
-            # transport limitations (sessions whose backing protocol cannot
-            # run PUT file transfers — write_pandas dies inside
-            # file_transfer_agent, e.g. KeyError('command')) both land here.
-            # Any failure falls back to the DDL-free SQL path, which is
-            # idempotent and carries all columns. Never leave the buffer
-            # poisoned: a raised exception here would re-fire on every
-            # subsequent flush-before-read and 500 the whole API.
-            logger.warning(
-                "pandas bulk path failed (%s: %s); falling back to SQL flush",
-                type(e).__name__,
-                e,
-            )
-            with contextlib.suppress(Exception):
-                self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
+            # requirement. Privilege denials (no CREATE TEMPORARY TABLE) are
+            # the EXPECTED steady state of least-privilege deployments —
+            # short-circuit quietly. Anything else (e.g. transports that
+            # cannot run PUT file transfers: write_pandas dies inside
+            # file_transfer_agent with KeyError('command')) is logged with
+            # the traceback, then falls back the same way. No cleanup DROP:
+            # temp tables die with the session, and a failure-path DROP is a
+            # doomed extra round trip that could even target a same-named
+            # permanent table when the temp CREATE never ran. A raised
+            # exception here would leave the buffer poisoned and re-fire on
+            # every subsequent flush-before-read (500ing the whole API), so
+            # every failure returns False. NOTE: a failure in the SQL
+            # fallback itself still propagates — flush error semantics
+            # beyond the pandas path are the caller's concern.
+            if not _is_privilege_error(e):
+                logger.warning(
+                    "pandas bulk path failed; falling back to SQL flush",
+                    exc_info=e,
+                )
             return False
         return True
 
@@ -390,27 +400,27 @@ def _flush_snapshots_pandas(self) -> bool:
         if not hasattr(conn, "_session_parameters"):
             return False
 
-        df = pd.DataFrame(
-            [
-                {
-                    "SNAPSHOT_HASH": s.snapshot_hash,
-                    "MODEL_HASH": s.model_hash,
-                    "PARENT_HASH": s.parent_hash,
-                    "TIMESTAMP": s.timestamp.isoformat(),
-                    "ACTOR": s.actor,
-                    "EVENT_TYPE": s.event_type,
-                    "SOURCE": s.source,
-                    "PAYLOAD": json.dumps(s.payload, default=str) if s.payload else None,
-                    "TAGS": json.dumps(s.tags, default=str) if s.tags else None,
-                }
-                for s in self._snapshot_buffer
-            ]
-        )
-
         staging = f"{self._schema}.SNAPSHOTS_STAGING"
         # See _flush_models_pandas: fall back to the DDL-free SQL path when the
-        # role can't create the staging table.
+        # role can't create the staging table. Row construction sits inside the
+        # try so a serialization surprise also falls back.
         try:
+            df = pd.DataFrame(
+                [
+                    {
+                        "SNAPSHOT_HASH": s.snapshot_hash,
+                        "MODEL_HASH": s.model_hash,
+                        "PARENT_HASH": s.parent_hash,
+                        "TIMESTAMP": s.timestamp.isoformat(),
+                        "ACTOR": s.actor,
+                        "EVENT_TYPE": s.event_type,
+                        "SOURCE": s.source,
+                        "PAYLOAD": json.dumps(s.payload, default=str) if s.payload else None,
+                        "TAGS": json.dumps(s.tags, default=str) if s.tags else None,
+                    }
+                    for s in self._snapshot_buffer
+                ]
+            )
             self._exec_no_result(
                 f"""
                 CREATE OR REPLACE TEMPORARY TABLE {staging} (
@@ -434,21 +444,25 @@ def _flush_snapshots_pandas(self) -> bool:
             self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
         except Exception as e:
             # The pandas path is an optimization, not a correctness
-            # requirement: privilege denials (no CREATE TEMPORARY TABLE) and
-            # transport limitations (sessions whose backing protocol cannot
-            # run PUT file transfers — write_pandas dies inside
-            # file_transfer_agent, e.g. KeyError('command')) both land here.
-            # Any failure falls back to the DDL-free SQL path, which is
-            # idempotent and carries all columns. Never leave the buffer
-            # poisoned: a raised exception here would re-fire on every
-            # subsequent flush-before-read and 500 the whole API.
-            logger.warning(
-                "pandas bulk path failed (%s: %s); falling back to SQL flush",
-                type(e).__name__,
-                e,
-            )
-            with contextlib.suppress(Exception):
-                self._exec_no_result(f"DROP TABLE IF EXISTS {staging}")
+            # requirement. Privilege denials (no CREATE TEMPORARY TABLE) are
+            # the EXPECTED steady state of least-privilege deployments —
+            # short-circuit quietly. Anything else (e.g. transports that
+            # cannot run PUT file transfers: write_pandas dies inside
+            # file_transfer_agent with KeyError('command')) is logged with
+            # the traceback, then falls back the same way. No cleanup DROP:
+            # temp tables die with the session, and a failure-path DROP is a
+            # doomed extra round trip that could even target a same-named
+            # permanent table when the temp CREATE never ran. A raised
+            # exception here would leave the buffer poisoned and re-fire on
+            # every subsequent flush-before-read (500ing the whole API), so
+            # every failure returns False. NOTE: a failure in the SQL
+            # fallback itself still propagates — flush error semantics
+            # beyond the pandas path are the caller's concern.
+            if not _is_privilege_error(e):
+                logger.warning(
+                    "pandas bulk path failed; falling back to SQL flush",
+                    exc_info=e,
+                )
             return False
         return True
 

tests/test_backends/test_snowflake_ledger.py

@@ -395,7 +395,11 @@ def sql(self, query: str, params: Any = None) -> MockCollectResult:
     backend.flush()
 
     assert len(inserts) == 1
-    row_source = inserts[0].split("FROM")[0]
+    # The row literals live in the UNION ALL source AFTER the first FROM —
+    # inspect that side (the original assertion checked the SELECT list and
+    # could never fail).
+    row_source = inserts[0].split("FROM", 1)[1]
+    assert "snap-nopayload-1" in row_source, "test is inspecting the wrong statement segment"
     assert "'null'" not in row_source.lower(), "empty payload serialized as JSON 'null' string"
 
 
@@ -1122,3 +1126,111 @@ def exploding_write_pandas(*args, **kwargs):
     before = len(statements)
     backend.flush()
     assert len(statements) == before
+
+
+def test_esc_escapes_backslashes_for_snowflake_literals():
+    """Snowflake processes backslash escapes inside single-quoted constants,
+    so JSON with embedded double quotes (json.dumps emits \\") or backslashes
+    must have backslashes doubled or PARSE_JSON receives mangled text —
+    verified live 2026-07-10: PARSE_JSON('{"a": "x \\" y"}') errors, the
+    doubled form round-trips.
+    """
+    import json as _json
+
+    from model_ledger.backends.snowflake import _esc
+
+    payload = {"summary": 'threshold "X" raised', "path": "dir\\file"}
+    literal = _esc(_json.dumps(payload))
+
+    inner = literal[1:-1].replace("''", "'")  # undo SQL quote doubling
+    unescaped = inner.replace("\\\\", "\\")  # what Snowflake's parser yields
+    assert _json.loads(unescaped) == payload, (
+        "literal does not round-trip through Snowflake unescaping"
+    )
+    assert '\\\\"' in literal, "embedded double quote not backslash-protected"
+
+
+def test_snapshot_payload_with_quotes_survives_sql_fallback():
+    """End-to-end through the fallback: a payload with quotes/backslashes must
+    produce an INSERT whose literals round-trip (the poisoned-buffer 500 loop
+    fired for ANY payload with an embedded double quote before the _esc fix).
+    """
+    import json as _json
+
+    from model_ledger.backends.snowflake import SnowflakeLedgerBackend
+
+    inserts: list[str] = []
+
+    class RecordingSession:
+        def sql(self, query: str, params: Any = None) -> MockCollectResult:
+            if "INSERT INTO" in query.upper() and ".SNAPSHOTS" in query.upper():
+                inserts.append(query)
+            return MockCollectResult([])
+
+    payload = {"observation": 'analyst wrote "backslash \\ and quote"', "nested": {"k": 'v"'}}
+    backend = SnowflakeLedgerBackend(schema="TEST_SCHEMA", connection=RecordingSession())
+    backend.append_snapshot(
+        Snapshot(
+            snapshot_hash="snap-esc-1",
+            model_hash="m1",
+            timestamp=datetime(2025, 1, 1, tzinfo=timezone.utc),
+            actor="scanner",
+            event_type="observation_issued",
+            source="alerting",
+            payload=payload,
+        )
+    )
+    backend.flush()
+
+    assert len(inserts) == 1
+    m = re.search(r"'(\{.*?\})'(?=,|\s)", inserts[0].split("FROM", 1)[1], re.DOTALL)
+    assert m, "payload literal not found in INSERT source"
+    snowflake_view = m.group(1).replace("''", "'").replace("\\\\", "\\")
+    assert _json.loads(snowflake_view) == payload
+
+
+def test_snapshot_pandas_failure_falls_back_to_sql(monkeypatch):
+    """The snapshots pandas path must fall back on non-privilege failures too
+    (the models-path twin of test_pandas_path_failure_falls_back_to_sql)."""
+    import sys
+    import types
+
+    from model_ledger.backends.snowflake import SnowflakeLedgerBackend
+
+    statements: list[str] = []
+
+    class FakeConn:
+        _session_parameters = {"QUERY_TAG": "test"}
+
+    class PandasCapableSession:
+        _connection = FakeConn()
+
+        def sql(self, query: str, params: Any = None) -> MockCollectResult:
+            statements.append(query)
+            return MockCollectResult([[0]])
+
+    def exploding_write_pandas(*args, **kwargs):
+        raise KeyError("command")
+
+    fake_tools = types.ModuleType("snowflake.connector.pandas_tools")
+    fake_tools.write_pandas = exploding_write_pandas
+    monkeypatch.setitem(sys.modules, "snowflake.connector.pandas_tools", fake_tools)
+
+    backend = SnowflakeLedgerBackend(schema="TEST_SCHEMA", connection=PandasCapableSession())
+    backend.append_snapshot(
+        Snapshot(
+            snapshot_hash="snap-fb-1",
+            model_hash="m1",
+            timestamp=datetime(2025, 1, 1, tzinfo=timezone.utc),
+            actor="scanner",
+            event_type="registered",
+            source="alerting",
+        )
+    )
+    backend.flush()  # must NOT raise
+
+    fallback_inserts = [
+        s for s in statements if "INSERT INTO" in s.upper() and ".SNAPSHOTS" in s.upper()
+    ]
+    assert fallback_inserts, "SQL fallback INSERT never ran after snapshots pandas-path failure"
+    assert not backend._snapshot_buffer, "snapshot buffer not cleared"