fix: enforce read-only execution in SQL example

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)

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