Harden MCP query_readonly so CTE-prefixed writes cannot mutate sessions

Harden MCP query_readonly so CTE-prefixed writes cannot mutate sessions

Executor instructions: Follow this issue step by step. Run every
verification command and confirm the expected result before moving on. If any
STOP condition is true, stop and report instead of improvising.

Drift check (run first): git diff --stat a11b6a3..HEAD -- src/plexosdb/db_manager.py src/plexosdb-mcp/src/plexosdb_mcp/server.py src/plexosdb-mcp/tests/test_mcp_server.py tests/test_db_manager.py
If any in-scope file changed since this issue was written, compare the
Current behavior excerpts against live code before proceeding. If they do not
match, stop and ask for /plan to refresh this issue.

Status

• Priority: P1
• Effort: S
• Risk: MED
• Depends on: none
• Category: security
• Planned at: commit a11b6a3, 2026-07-08

Current behavior

The new MCP server exposes a tool named query_readonly, and the docs recommend
read-only mode for untrusted prompts/hosts. The current SQL gate only checks the
first characters of the submitted SQL. It accepts any statement beginning with
with, then delegates to PlexosDB.query() / DatabaseManager.query().

• src/plexosdb-mcp/src/plexosdb_mcp/server.py:625-638 — MCP tool-level gate accepts select or with.

# src/plexosdb-mcp/src/plexosdb_mcp/server.py:625-638
def query_readonly(
    session_id: str,
    sql: str,
    params: tuple[Any, ...] | dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Execute a read-only SQL query (SELECT/CTE) and return rows."""
    statement = sql.lstrip().lower()
    if not (statement.startswith("select") or statement.startswith("with")):
        raise ValueError("query_readonly only allows SELECT/CTE statements")

    db = server_state.get_db(session_id)
    rows = db.query(sql, params=params)

• src/plexosdb/db_manager.py:448-455 — core query validation rejects only first-word write keywords.

# src/plexosdb/db_manager.py:448-455
def _validate_query_type(self, query: str) -> None:
    """Validate that query is read-only for query methods."""
    query_upper = query.strip().upper()
    write_keywords = {"INSERT", "UPDATE", "DELETE", "DROP", "CREATE", "ALTER"}
    first_word = query_upper.split()[0] if query_upper.split() else ""

    if first_word in write_keywords:
        raise ValueError(f"Use execute() for {first_word} statements, not query()")

• src/plexosdb-mcp/tests/test_mcp_server.py:587-600 — coverage rejects direct write statements, but not write-capable CTE statements.

# src/plexosdb-mcp/tests/test_mcp_server.py:587-600
def test_tool_input_validation_errors(monkeypatch: pytest.MonkeyPatch) -> None:
    """Invalid class/collection names and unsafe SQL queries raise ValueError."""
    monkeypatch.setattr(mcp_server, "FastMCP", FakeFastMCP)
    state = FakeState()
    mcp = mcp_server.build_mcp_server(state)

    with pytest.raises(ValueError, match="Invalid class_name"):
        _ = mcp.tools["list_objects_by_class"]("sid", "NotAClass")

    with pytest.raises(ValueError, match="only allows SELECT/CTE"):
        _ = mcp.tools["query_readonly"]("sid", "DELETE FROM t_object")

Desired behavior or Goal

query_readonly and the underlying PlexosDB.query() path must enforce a true
read-only contract. Plain SELECT queries and read-only CTE queries continue to
work. Any statement whose execution would create, update, delete, attach, detach,
replace, vacuum, or otherwise mutate database state is rejected before mutation.
This protects MCP hosts that expose query_readonly to agents or untrusted prompt
content.

Acceptance criteria

• PlexosDB.query() / DatabaseManager.query() still accepts normal SELECT statements.
• Read-only CTE statements whose top-level operation is SELECT still work when they worked before this change.
• Write-capable CTE statements are rejected before any table contents change.
• MCP query_readonly uses the same read-only enforcement as the core DB query path or delegates to a core helper that enforces it.
• Existing MCP response shapes for successful read queries remain unchanged: session_id, count, and rows are still returned.
• Regression tests cover both direct write statements and CTE-prefixed write statements.

Non-goals

• Do not add a general SQL execution tool to the MCP server.
• Do not change the behavior or names of edit/export MCP tools such as add_object, delete_object, save_xml, or to_csv.
• Do not expose additional database schema details in error messages; keep errors concise and safe for agent-facing output.
• Do not broaden this issue into a full SQL parser rewrite beyond the read-only enforcement needed by query() / query_readonly.

Work Plan

Validation

• uv run pytest tests/test_db_manager.py -k "query" exits 0 and includes the new read-only enforcement cases.
• uv run --project src/plexosdb-mcp pytest -q -c src/plexosdb-mcp/pyproject.toml src/plexosdb-mcp/tests/test_mcp_server.py -k "query_readonly or input_validation" exits 0 and covers MCP rejection of CTE-prefixed writes.
• uv run ty check --output-format github ./src/plexosdb exits 0.
• uv run prek run --show-diff-on-failure --color=always --all-files --hook-stage pre-push exits 0 before handoff.

Documentation

• None for the first fix. The existing public contract already says the tool is read-only; this issue makes the implementation match that contract.

Testing

• Add focused tests in tests/test_db_manager.py for direct write rejection, read-only SELECT acceptance, read-only CTE acceptance, and CTE-prefixed write rejection.
• Add a focused MCP tool test in src/plexosdb-mcp/tests/test_mcp_server.py proving query_readonly rejects a CTE-prefixed write and leaves the fake/real target state unchanged.
• Model MCP tests after the existing test_tool_input_validation_errors style in src/plexosdb-mcp/tests/test_mcp_server.py.

Risks

Breaking-change

Low to medium. The documented query contract says SELECT/read-only queries only, so rejecting write-capable SQL is intended. The risk is accidentally rejecting legitimate read-only CTEs or query forms currently used by maintainers.

Review-size

Low. Scope is limited to the core query validation path plus focused tests in the root package and MCP package.

Implementation notes

Scope

In scope:

• src/plexosdb/db_manager.py
• tests/test_db_manager.py
• src/plexosdb-mcp/src/plexosdb_mcp/server.py
• src/plexosdb-mcp/tests/test_mcp_server.py

Out of scope:

• src/plexosdb/db.py except for type/docstring adjustments directly required by the query helper signature.
• src/plexosdb/schema.sql because this issue does not change schema.
• docs/source/howtos/mcp_server.md because the docs already describe the intended read-only behavior.

Suggested steps

1. Write failing regression tests first.
   • In tests/test_db_manager.py, create a small in-memory table through the existing DB manager setup pattern, then assert direct writes and CTE-prefixed writes through query() raise and do not change row counts.
   • In src/plexosdb-mcp/tests/test_mcp_server.py, add an MCP-level rejection test for a CTE-prefixed write.
   • Verify: the focused pytest commands above fail before implementation for the new CTE-prefixed write case.
2. Strengthen the core read-only enforcement in src/plexosdb/db_manager.py.
   • Prefer SQLite-backed enforcement such as a temporary authorizer or an equivalent mechanism that denies mutation opcodes during query() execution.
   • Keep the cheap first-token allowlist for obvious non-read statements, but do not rely on it as the only protection for CTEs.
   • Deny mutation categories including insert, update, delete, create, drop, alter, attach, detach, replace, vacuum, and write-capable pragma behavior.
3. Make query_readonly delegate to the core enforcement instead of maintaining a weaker MCP-only SQL classifier.
   • The MCP tool may keep a small user-friendly precheck for SELECT / WITH, but the core query path must be the authority.
4. Run focused validation, then the broad pre-push gate.

Test details

• Use existing root DB manager fixture/style from tests/test_db_manager.py.
• Use the existing FakeFastMCP / FakeState pattern in src/plexosdb-mcp/tests/test_mcp_server.py for MCP registration tests.
• The regression must assert state remains unchanged after a rejected statement, not only that an exception was raised.

Maintenance notes

• Future MCP read/query tools must call the same core read-only enforcement helper.
• Reviewers should scrutinize whether the protection is enforced at execution time, not only through string matching.

Start with the current validation in src/plexosdb/db_manager.py and the query_readonly entry point in src/plexosdb-mcp/src/plexosdb_mcp/server.py. Run the focused pytest commands from the Validation section, then add regression coverage in tests/test_db_manager.py and src/plexosdb-mcp/tests/test_mcp_server.py. Done means SELECT and read-only CTE queries still work, write-capable CTEs are rejected before state changes, and MCP response shapes remain unchanged.