From 01678c90bd50622b80d6a0371a67c4d3600df8fe Mon Sep 17 00:00:00 2001 From: David Fitzsimmons Date: Sat, 5 Sep 2026 09:44:23 -0400 Subject: [PATCH] fix: enforce read-only SQL example verification --- examples/04_sql_synth.py | 15 ++++++-- tests/test_sql_example.py | 75 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 2 deletions(-) create mode 100644 tests/test_sql_example.py diff --git a/examples/04_sql_synth.py b/examples/04_sql_synth.py index 9e9912c..ab4f0a9 100644 --- a/examples/04_sql_synth.py +++ b/examples/04_sql_synth.py @@ -4,8 +4,8 @@ 1. Claude is given a natural-language query against a small in-memory schema (employees, departments). 2. Claude returns a single SELECT statement. - 3. We refuse anything that isn't a single SELECT, execute it against - the fixture, and compare its result set to the expected set. + 3. SQLite permits only read operations for the candidate. We execute it + against the fixture and compare its result set to the expected set. 4. Error = size of the symmetric difference of result rows. 0 fires TARGET_MET. @@ -80,6 +80,15 @@ def extract_sql(text: str) -> str: return (fence.group(1) if fence else text).strip().rstrip(";").strip() +def _readonly_authorizer(action, _arg1, arg2, _database, _source): + """Permit reads only; statement prefixes are not a security boundary.""" + if action == sqlite3.SQLITE_FUNCTION: + return sqlite3.SQLITE_DENY if arg2.lower() == "load_extension" else sqlite3.SQLITE_OK + if action in (sqlite3.SQLITE_SELECT, sqlite3.SQLITE_READ, sqlite3.SQLITE_RECURSIVE): + return sqlite3.SQLITE_OK + return sqlite3.SQLITE_DENY + + def run_query(conn: sqlite3.Connection, sql: str): if not sql: return len(EXPECTED) + 1, "empty query" @@ -88,6 +97,7 @@ def run_query(conn: sqlite3.Connection, sql: str): if ";" in sql: return len(EXPECTED) + 1, "multi-statement input rejected" try: + conn.set_authorizer(_readonly_authorizer) rows = conn.execute(sql).fetchall() except sqlite3.Error as exc: return len(EXPECTED) + 1, f"sql error: {exc}" @@ -143,6 +153,7 @@ def main() -> None: client = get_client() conn = sqlite3.connect(":memory:") conn.executescript(SCHEMA) + conn.execute("PRAGMA query_only=ON") print("Spec: top-2-per-department over an 8-row fixture.\n") baseline_err, baseline_iters = baseline_run(client, conn) lg = loopgain_run(client, conn) diff --git a/tests/test_sql_example.py b/tests/test_sql_example.py new file mode 100644 index 0000000..aad5e55 --- /dev/null +++ b/tests/test_sql_example.py @@ -0,0 +1,75 @@ +"""Offline checks for the SQL example's in-memory verifier.""" +import importlib.util +from pathlib import Path +import sqlite3 + +import pytest + + +@pytest.fixture +def example(monkeypatch): + directory = Path(__file__).resolve().parents[1] / "examples" + monkeypatch.syspath_prepend(str(directory)) + spec = importlib.util.spec_from_file_location("sql_example", directory / "04_sql_synth.py") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture +def database(example): + with sqlite3.connect(":memory:") as conn: + conn.executescript(example.SCHEMA) + yield conn + + +@pytest.mark.parametrize("sql", [ + "SELECT name FROM employees", + "WITH names AS (SELECT name FROM employees) SELECT name FROM names", + "WITH RECURSIVE nums(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM nums WHERE n<3) SELECT n FROM nums", +]) +def test_read_queries_remain_supported(example, database, sql): + error, message = example.run_query(database, sql) + assert error >= 0 + assert "sql error" not in message + assert "rows;" in message + + +def test_expected_window_query_still_scores_zero(example, database): + sql = """WITH ranked AS ( + SELECT d.name AS department_name, e.name AS employee_name, e.salary, + ROW_NUMBER() OVER (PARTITION BY d.id ORDER BY e.salary DESC) AS rank + FROM employees e JOIN departments d ON e.department_id = d.id + ) SELECT department_name, employee_name, salary FROM ranked WHERE rank <= 2 + ORDER BY department_name, salary DESC""" + assert example.run_query(database, sql)[0] == 0 + assert example.run_query(database, sql)[0] == 0 + + +@pytest.mark.parametrize("sql", [ + "WITH marker AS (SELECT 1) DELETE FROM employees", + "WITH marker AS (SELECT 1) UPDATE employees SET salary = 0", + "WITH marker AS (SELECT 1) INSERT INTO employees VALUES (9, 'Synthetic', 1, 0)", +]) +def test_mutations_cannot_change_fixture(example, database, sql): + before = database.execute("SELECT * FROM employees ORDER BY id").fetchall() + error, message = example.run_query(database, sql) + assert error > 0 + assert "sql error" in message + assert database.execute("SELECT * FROM employees ORDER BY id").fetchall() == before + assert "rows;" in example.run_query(database, "SELECT name FROM employees")[1] + + +def test_runtime_guard_rejects_nonread_operations(example, database): + example.run_query(database, "SELECT name FROM employees") + for statement in ["DELETE FROM employees", "CREATE TABLE extra(id)", + "ATTACH DATABASE ':memory:' AS extra", "PRAGMA query_only=OFF"]: + with pytest.raises(sqlite3.DatabaseError): + database.execute(statement) + + +def test_extension_loading_is_denied_by_authorizer(example, database): + # The authorizer denies the function before the extension loader is invoked. + error, message = example.run_query(database, "SELECT load_extension('synthetic-not-a-file')") + assert error > 0 + assert "not authorized" in message