diff --git a/pyproject.toml b/pyproject.toml index 6e04056..31cc59a 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/model_ledger/backends/snowflake.py b/src/model_ledger/backends/snowflake.py index 972c10c..2805f6f 100644 --- a/src/model_ledger/backends/snowflake.py +++ b/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 diff --git a/tests/test_backends/test_snowflake_ledger.py b/tests/test_backends/test_snowflake_ledger.py index 8130cba..1e5bd23 100644 --- a/tests/test_backends/test_snowflake_ledger.py +++ b/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"