feat: Add readonly access mode for database-level read-only enforcement

[yaryrslv]

Summary

Add a third access mode --access-mode=readonly that enforces read-only
transactions at the PostgreSQL level without pglast SQL validation.
This fills the gap between unrestricted (no protection) and restricted
(pglast rejects complex queries).

Problem

The restricted mode uses pglast to parse SQL into an AST and validate
statements against whitelists. While this catches writes at the application
layer, it also rejects legitimate read-only queries that pglast cannot
parse:

• PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY duration) — pglast parse error
• Nested CTEs with complex joins — rejected as unsupported node types
• Window functions with advanced framing — pglast validation failure
• Extension-specific functions (pg_trgm, postgis, etc.) — not in the allowed function whitelist

Users running analytical workloads are forced to switch to unrestricted
mode, losing all write protection.

Solution

Architecture: Driver selection

The server selects a driver based on --access-mode CLI argument:

flowchart TD
    CLI["--access-mode=?"] --> U{unrestricted}
    CLI --> RO{readonly}
    CLI --> R{restricted}

    U --> SD["SqlDriver\n(no wrapper)"]
    RO --> ROSD["ReadOnlySqlDriver\n(decorator over SqlDriver)"]
    R --> SSD["SafeSqlDriver\n(decorator over SqlDriver)"]

    SD --> EQ["execute_query()\nforce_readonly=False"]
    ROSD --> EQ2["execute_query()\nforce_readonly=True ← forced\n+ /* crystaldba */ prefix\n+ 30s timeout"]
    SSD --> VAL["_validate(query)\npglast AST parsing"]
    VAL -->|pass| EQ3["execute_query()\nforce_readonly=True ← forced\n+ /* crystaldba */ prefix\n+ 30s timeout"]
    VAL -->|fail| ERR["Reject query\n(ValueError)"]

    EQ --> PG[(PostgreSQL)]
    EQ2 --> PG
    EQ3 --> PG

    style ROSD fill:#2d6a4f,stroke:#1b4332,color:#fff
    style RO fill:#2d6a4f,stroke:#1b4332,color:#fff

Loading

Architecture: ReadOnlySqlDriver decorator pattern

flowchart TD
    CALLER["MCP Tool calls execute_query()"] --> ROSD

    subgraph ROSD["ReadOnlySqlDriver"]
        A["Ignore caller's force_readonly value"] --> B["Prepend /* crystaldba */ comment"]
        B --> C["Wrap in asyncio.timeout(30s)"]
        C --> D["Call self.sql_driver.execute_query(\n  query='/* crystaldba */ ...',\n  params=params,\n  force_readonly=True\n)"]
    end

    D --> EXEC

    subgraph EXEC["SqlDriver._execute_with_connection"]
        E["BEGIN TRANSACTION READ ONLY"] --> F["Execute user's query"]
        F --> G{Success?}
        G -->|Yes| H["ROLLBACK"]
        G -->|No| I["connection.rollback()"]
        H --> J["Return results"]
        I --> K["Raise exception"]
    end

    J --> RESULT["MCP Tool Response\n(query results)"]
    K --> ERROR["MCP Tool Response\n(error message)"]

    style ROSD fill:#2d6a4f,stroke:#1b4332,color:#fff
    style EXEC fill:#1b4332,stroke:#081c15,color:#fff

Loading

Architecture: Query flow in readonly mode (sequence)

sequenceDiagram
    participant Tool as MCP Tool
    participant RO as ReadOnlySqlDriver
    participant SD as SqlDriver
    participant PG as PostgreSQL

    Tool->>RO: execute_query(sql, params, force_readonly=False)
    Note over RO: Ignores force_readonly=False
    Note over RO: Prepends /* crystaldba */
    Note over RO: Starts asyncio.timeout(30s)

    RO->>SD: execute_query("/* crystaldba */ " + sql, params, force_readonly=True)
    SD->>PG: BEGIN TRANSACTION READ ONLY
    PG-->>SD: OK
    SD->>PG: /* crystaldba */ SELECT ...

    alt Query succeeds
        PG-->>SD: rows
        SD->>PG: ROLLBACK
        SD-->>RO: List[RowResult]
        RO-->>Tool: List[RowResult]
    else Write attempt (INSERT, UPDATE, DROP...)
        PG-->>SD: ERROR: cannot execute INSERT in a read-only transaction
        SD->>PG: ROLLBACK
        SD-->>RO: raises Exception
        RO-->>Tool: raises Exception
    else Timeout (>30s)
        Note over RO: asyncio.TimeoutError
        RO-->>Tool: ValueError("Query execution timed out after 30 seconds")
    end

Loading

Architecture: Protection layers comparison

graph LR
    subgraph unrestricted["unrestricted mode"]
        U1["SQL query"] --> U2["SqlDriver"] --> U3[(PostgreSQL)]
    end

    subgraph readonly["readonly mode"]
        R1["SQL query"] --> R2["ReadOnlySqlDriver\n(timeout + prefix)"] --> R3["SqlDriver\nforce_readonly=True"] --> R4[("PostgreSQL\nREAD ONLY tx")]
    end

    subgraph restricted["restricted mode"]
        S1["SQL query"] --> S2["SafeSqlDriver\npglast validation"] --> S3["SqlDriver\nforce_readonly=True"] --> S4[("PostgreSQL\nREAD ONLY tx")]
    end

    style unrestricted fill:#d4380d,stroke:#a8071a,color:#fff
    style readonly fill:#2d6a4f,stroke:#1b4332,color:#fff
    style restricted fill:#0050b3,stroke:#003a8c,color:#fff

Loading

Mode comparison

	unrestricted	readonly	restricted
SQL validation	none	none	pglast AST check
DB enforcement	none	READ ONLY transaction	READ ONLY transaction
Complex queries	all allowed	all allowed	limited by pglast
Timeout	none	30 seconds	30 seconds
Query prefix	none	/* crystaldba */	/* crystaldba */
Protection level	none	database-level	dual-layer (app + DB)
Best for	development	analytics / complex queries	production

Security trade-off

flowchart LR
    subgraph restricted_mode["restricted mode (dual-layer)"]
        direction TB
        L1["Layer 1: pglast\nAST validation"] --> L2["Layer 2: PostgreSQL\nREAD ONLY transaction"]
    end

    subgraph readonly_mode["readonly mode (single-layer)"]
        direction TB
        L3["No SQL validation"] --> L4["PostgreSQL\nREAD ONLY transaction"]
    end

    ATK["Multi-statement:\nCOMMIT; DROP TABLE users"]

    ATK -->|blocked by pglast| restricted_mode
    ATK -->|not parsed, but\nDROP fails in\nREAD ONLY tx| readonly_mode

    style restricted_mode fill:#0050b3,stroke:#003a8c,color:#fff
    style readonly_mode fill:#2d6a4f,stroke:#1b4332,color:#fff

Loading

Readonly mode does not validate SQL syntax. Multi-statement payloads
like COMMIT; DROP TABLE users are not caught at the application layer.
PostgreSQL's read-only transaction prevents the write from executing,
but this is single-layer protection vs. the dual-layer approach of
restricted mode.

This trade-off is intentional: pglast's false rejections of valid
analytical queries cause more real-world friction than the multi-statement
attack vector, which is mitigated by the database transaction boundary.

Usage

# Start server in readonly mode
postgres-mcp "postgresql://user:pass@host/db" --access-mode=readonly

# Start server in restricted mode (strictest)
postgres-mcp "postgresql://user:pass@host/db" --access-mode=restricted

# Start server in unrestricted mode (default)
postgres-mcp "postgresql://user:pass@host/db" --access-mode=unrestricted

Implementation details

ReadOnlySqlDriver class (src/postgres_mcp/sql/readonly_sql.py)

• Pattern: Decorator over SqlDriver
• Constructor: ReadOnlySqlDriver(sql_driver: SqlDriver, timeout: float | None = None)
• Key behavior: execute_query() always passes force_readonly=True to the underlying driver, regardless of the caller's value
• Timeout: Wraps execution in asyncio.timeout(self.timeout), raises ValueError on expiry
• Query tagging: Prepends /* crystaldba */ comment for identification in pg_stat_activity and logs
• No SQL validation: Unlike SafeSqlDriver, does not call _validate() / pglast

get_sql_driver() dispatch (src/postgres_mcp/server.py)

async def get_sql_driver():
    base_driver = SqlDriver(conn=db_connection)
    if current_access_mode == AccessMode.RESTRICTED:
        return SafeSqlDriver(sql_driver=base_driver, timeout=30)
    elif current_access_mode == AccessMode.READONLY:
        return ReadOnlySqlDriver(sql_driver=base_driver, timeout=30)
    else:
        return base_driver

Transaction enforcement (src/postgres_mcp/sql/sql_driver.py)

When force_readonly=True, _execute_with_connection() wraps every query:

BEGIN TRANSACTION READ ONLY   -- PostgreSQL enforces no writes
/* crystaldba */ <query>      -- user's query runs here
ROLLBACK                      -- transaction closed (no state change)

PostgreSQL rejects any INSERT, UPDATE, DELETE, DROP, CREATE,
TRUNCATE, or other write operation inside a READ ONLY transaction with:

ERROR: cannot execute <STATEMENT> in a read-only transaction

Files changed

File	Change
src/postgres_mcp/sql/readonly_sql.py	New ReadOnlySqlDriver class (decorator, timeout, force_readonly)
src/postgres_mcp/server.py	AccessMode.READONLY enum value, get_sql_driver() branch, CLI --access-mode=readonly
src/postgres_mcp/sql/__init__.py	Export ReadOnlySqlDriver
smithery.yaml	Add readonly to access mode config
README.md	Update "Protected SQL Execution" docs (two → three levels)
tests/unit/sql/test_readonly_sql.py	8 unit tests for ReadOnlySqlDriver
tests/unit/sql/test_readonly_enforcement.py	3 parameterized tests — force_readonly enforcement across all access modes
tests/unit/test_access_mode.py	Driver selection + CLI parsing for readonly

Test coverage

• test_readonly_sql.py (8 tests): force_readonly override, no SQL validation,
  /* crystaldba */ prefix, timeout → ValueError, parameter forwarding,
  exception propagation, None result handling
• test_readonly_enforcement.py (3 parameterized tests): verifies force_readonly
  behavior per access mode (unrestricted respects caller, readonly/restricted force True)
• test_access_mode.py: get_sql_driver() returns correct type per mode,
  timeout=30 for restricted/readonly, CLI --access-mode parsing