From 7cc9fecbdcbb987cacdc720583f73c3aed960851 Mon Sep 17 00:00:00 2001 From: plx Date: Sun, 16 Aug 2026 00:59:09 +0000 Subject: [PATCH] fix: establish read-only audit transaction before identity check --- scripts/simulate_blif_flow_v2.py | 33 +++-- ...blif_flow_v2_readonly_transaction_order.py | 125 ++++++++++++++++++ 2 files changed, 149 insertions(+), 9 deletions(-) create mode 100644 tests/test_blif_flow_v2_readonly_transaction_order.py diff --git a/scripts/simulate_blif_flow_v2.py b/scripts/simulate_blif_flow_v2.py index 302fa7b..7d6ec6e 100644 --- a/scripts/simulate_blif_flow_v2.py +++ b/scripts/simulate_blif_flow_v2.py @@ -99,15 +99,29 @@ def _load( ) -> dict[str, Any]: with engine.connect() as conn: conn = conn.execution_options(isolation_level="AUTOCOMMIT") - identity = conn.execute(text( - "SELECT current_database(), current_user, current_setting('transaction_read_only')" - )).one() - if identity[0] != expected_database or (expected_user and identity[1] != expected_user): - raise RuntimeError(f"refusing unexpected database identity: {identity!r}") - if require_read_only and identity[2] != "on": - raise RuntimeError(f"read-only simulation requires transaction_read_only=on: {identity!r}") - conn.execute(text("BEGIN READ ONLY")) + transaction_started = False try: + # A fresh connection commonly reports transaction_read_only=off. + # Establish the protected transaction on the same connection used + # for every factual read before validating a strict audit. + if require_read_only: + conn.execute(text("BEGIN READ ONLY")) + transaction_started = True + + identity = conn.execute(text( + "SELECT current_database(), current_user, current_setting('transaction_read_only')" + )).one() + if identity[0] != expected_database or (expected_user and identity[1] != expected_user): + raise RuntimeError(f"refusing unexpected database identity: {identity!r}") + if require_read_only and identity[2] != "on": + raise RuntimeError(f"read-only simulation requires transaction_read_only=on: {identity!r}") + + # Preserve the development/test path's established ordering: check + # its identity first, then protect the factual reads themselves. + if not require_read_only: + conn.execute(text("BEGIN READ ONLY")) + transaction_started = True + opportunities = [dict(row) for row in conn.execute(text(""" SELECT o.*, o.id::text AS id, o.local_customer_id::text, c.name AS linked_customer_name, c.tax_id, c.email AS fiscal_email, @@ -157,7 +171,8 @@ def _load( WHERE status IN ('open','needs_review','conflict') ORDER BY created_at """)).mappings()] finally: - conn.execute(text("ROLLBACK")) + if transaction_started: + conn.execute(text("ROLLBACK")) return { "identity": {"database": identity[0], "user": identity[1], "transaction_read_only": identity[2]}, "opportunities": opportunities, "tasks": _group(tasks), "messages": _group(messages), diff --git a/tests/test_blif_flow_v2_readonly_transaction_order.py b/tests/test_blif_flow_v2_readonly_transaction_order.py new file mode 100644 index 0000000..0e239e3 --- /dev/null +++ b/tests/test_blif_flow_v2_readonly_transaction_order.py @@ -0,0 +1,125 @@ +from contextlib import nullcontext + +import pytest + +import scripts.simulate_blif_flow_v2 as simulator + + +class _Result: + def __init__(self, *, identity=None, rows=()): + self._identity = identity + self._rows = rows + + def one(self): + return self._identity + + def mappings(self): + return self._rows + + +class _Connection: + def __init__(self, database="clientflow", user="clientflow"): + self.database = database + self.user = user + self.read_only = False + self.statements = [] + self.identity_observations = [] + + def execution_options(self, **_kwargs): + return self + + def execute(self, statement): + sql = " ".join(str(statement).split()) + self.statements.append(sql) + upper = sql.upper() + if upper == "BEGIN READ ONLY": + self.read_only = True + return _Result() + if upper == "ROLLBACK": + self.read_only = False + return _Result() + if upper.startswith(("INSERT ", "UPDATE ", "DELETE ")): + if self.read_only: + raise RuntimeError("cannot execute write in a read-only transaction") + return _Result() + if "CURRENT_DATABASE()" in upper: + identity = (self.database, self.user, "on" if self.read_only else "off") + self.identity_observations.append(identity) + return _Result(identity=identity) + return _Result(rows=()) + + +class _Engine: + def __init__(self, connection): + self.connection = connection + + def connect(self): + return nullcontext(self.connection) + + +def _load(monkeypatch, connection, **kwargs): + monkeypatch.setattr(simulator, "engine", _Engine(connection)) + return simulator._load(**kwargs) + + +def test_strict_load_establishes_read_only_before_same_connection_identity_and_reads(monkeypatch): + connection = _Connection() + assert connection.read_only is False # normal fresh-connection state + + result = _load( + monkeypatch, connection, expected_database="clientflow", + expected_user="clientflow", require_read_only=True, + ) + + assert connection.statements[0] == "BEGIN READ ONLY" + assert "CURRENT_DATABASE()" in connection.statements[1].upper() + assert connection.identity_observations == [("clientflow", "clientflow", "on")] + assert result["identity"]["transaction_read_only"] == "on" + assert connection.statements[2].upper().startswith("SELECT O.*") + assert connection.statements[-1] == "ROLLBACK" + + +@pytest.mark.parametrize( + ("database", "user", "expected_database", "expected_user"), + [ + ("wrong", "clientflow", "clientflow", "clientflow"), + ("clientflow", "wrong", "clientflow", "clientflow"), + ], +) +def test_strict_load_refuses_wrong_identity_before_factual_reads( + monkeypatch, database, user, expected_database, expected_user, +): + connection = _Connection(database=database, user=user) + with pytest.raises(RuntimeError, match="unexpected database identity"): + _load( + monkeypatch, connection, expected_database=expected_database, + expected_user=expected_user, require_read_only=True, + ) + + assert connection.statements[0] == "BEGIN READ ONLY" + assert len([sql for sql in connection.statements if sql.upper().startswith("SELECT")]) == 1 + assert connection.statements[-1] == "ROLLBACK" + + +def test_strict_factual_transaction_rejects_writes(): + connection = _Connection() + connection.execute("BEGIN READ ONLY") + with pytest.raises(RuntimeError, match="read-only transaction"): + connection.execute("UPDATE opportunities SET stage = 'forbidden'") + assert connection.read_only is True + + +def test_non_strict_load_preserves_identity_then_read_only_collection_order(monkeypatch): + connection = _Connection(database="clientflow_codex_test", user="clientflow_codex_test") + result = _load( + monkeypatch, connection, expected_database="clientflow_codex_test", + expected_user="clientflow_codex_test", require_read_only=False, + ) + + assert "CURRENT_DATABASE()" in connection.statements[0].upper() + assert connection.identity_observations == [ + ("clientflow_codex_test", "clientflow_codex_test", "off") + ] + assert connection.statements[1] == "BEGIN READ ONLY" + assert result["identity"]["transaction_read_only"] == "off" + assert connection.statements[-1] == "ROLLBACK"