feat(sql): first-party SQL-safety kernel, proven at parity (18.1, PR 1)

[devopam]

Summary

PR 1 of the de-vendor effort (plan, roadmap 18.1). Additive — lands a first-party mcpg.sql package and proves it's behaviourally identical to the vendored kernel, while _vendor/ stays the live path. No consumer swings yet (that's PR 2), so no snapshot changes.

New package src/mcpg/sql/

• driver.py — SqlDriver / DbConnPool / obfuscate_password, re-authored from the vendored sql_driver.py. Behaviour identical, typing modernised, zero SQL-safety policy.
• allowlist.py — the SQL-safety policy as data: the permitted statement / AST-node / function / extension sets + patterns, extracted verbatim (the single auditable decision surface).
• safety.py — SafeSqlDriver: the pglast parse + AST-walker + execute path, reproduced faithfully, reading policy from allowlist.py (mechanism separated from policy; the walker can't widen what's allowed). Class attributes alias the module constants so SafeSqlDriver.ALLOWED_* still resolves.

This is the re-architecture from the plan — the vendored 1,036-LOC monolith split into policy (data) + mechanism.

Faithful re-author, proven

• Adversarial suite (test_sql_kernel_safety.py, ported from the 760-LOC tests/vendor/sql/test_safe_sql.py, import-swap only) — 60/60 pass against mcpg.sql.
• Differential parity harness (test_sql_kernel_differential.py) — a 48-input safe/unsafe/malformed corpus through both validators, asserting an identical accept/reject verdict + error type. 50 cases, zero divergence. This is the security-review gate's core evidence; it's deleted in PR 2 with _vendor/.
• Driver + obfuscate tests ported too. 125 kernel tests green.

Gates

The new code is inside ruff + mypy --strict from day one — the vendored allowlist never was. (The full _vendor carve-out removal + coverage fold-in happens in PR 2 when _vendor/ is deleted.)

Note

The /* crystaldba */ execution marker is kept verbatim (the adversarial suite asserts on it, and parity is the priority). A cosmetic rename to /* mcpg */ can follow once the swing is locked in.

Roadmap linkage

Advances roadmap row: 18.1 (PR 1 of 2; the swing + delete + gate-removal is PR 2).

Checklist

• Tests first / ported; adversarial suite + differential parity green (125 tests)
• ruff, ruff format, mypy --strict pass on the new kernel
• Purely additive — _vendor/ untouched, tool snapshots unchanged
• No hand-edits to src/mcpg/_vendor/

🤖 Generated with Claude Code

---

Generated by Claude Code

[sourcery-ai]

Sorry @devopam, you have reached your weekly rate limit of 500000 diff characters.

Please try again later or upgrade to continue using Sourcery

[gemini-code-assist-2]

Code Review

This pull request replaces the vendored SQL kernel with a first-party implementation containing a connection pool, query driver, and SQL-safety allowlist mechanism. While the architecture is solid, several critical issues in src/mcpg/sql/driver.py need to be resolved. Specifically, obfuscate_password fails to redact URL-encoded passwords, concurrent pool initialization is prone to race conditions and resource leaks, direct connections are leaked on failure, and transaction handling in _execute_with_connection bypasses psycopg3's internal state tracking while failing to roll back on errors.

[gemini-code-assist-2]

The current implementation of obfuscate_password fails to redact passwords that contain URL-encoded special characters (e.g., %40 for @).

When urlparse parses the URL, parsed.password returns the decoded password, whereas parsed.netloc contains the raw/encoded password. As a result, parsed.netloc.replace(parsed.password, "****") fails to find a match, leaking the credential in logs and error messages.

We should rebuild the netloc safely by splitting the userinfo and host parts to ensure both plain and URL-encoded passwords are consistently obfuscated.

Suggested change

	# Try first as a proper URL.
	try:
	parsed = urlparse(text)
	if parsed.scheme and parsed.netloc and parsed.password:
	netloc = parsed.netloc.replace(parsed.password, "****")
	return urlunparse(parsed._replace(netloc=netloc))
	except Exception:
	pass
	# Try first as a proper URL.
	try:
	parsed = urlparse(text)
	if parsed.scheme and parsed.netloc and parsed.password:
	if "@" in parsed.netloc:
	user_pass, host_port = parsed.netloc.rsplit("@", 1)
	if ":" in user_pass:
	user, _ = user_pass.split(":", 1)
	netloc = f"{user}:****@{host_port}"
	else:
	netloc = f"{user_pass}:****@{host_port}"
	else:
	netloc = parsed.netloc
	return urlunparse(parsed._replace(netloc=netloc))
	except Exception:
	pass

[gemini-code-assist-2]

Initialize an asyncio.Lock to serialize connection pool creation and prevent race conditions when multiple concurrent tasks attempt to initialize the pool simultaneously.

Suggested change

	def __init__(self, connection_url: str | None = None, min_size: int = 1, max_size: int = 5) -> None:
	self.connection_url = connection_url
	self.min_size = min_size
	self.max_size = max_size
	self.pool: AsyncConnectionPool | None = None
	self._is_valid = False
	self._last_error: str | None = None
	def __init__(self, connection_url: str | None = None, min_size: int = 1, max_size: int = 5) -> None:
	import asyncio
	self.connection_url = connection_url
	self.min_size = min_size
	self.max_size = max_size
	self.pool: AsyncConnectionPool | None = None
	self._is_valid = False
	self._last_error: str | None = None
	self._lock = asyncio.Lock()

[gemini-code-assist-2]

Use the initialized _lock to serialize pool connection attempts. This prevents concurrent tasks from concurrently calling close(), creating multiple pools, and overwriting self.pool, which leads to resource leaks and connection errors.

Suggested change

	async def pool_connect(self, connection_url: str | None = None) -> AsyncConnectionPool:
	"""Open (once) and return the pool, testing it with ``SELECT 1``."""
	if self.pool and self._is_valid:
	return self.pool
	async def pool_connect(self, connection_url: str | None = None) -> AsyncConnectionPool:
	"""Open (once) and return the pool, testing it with ``SELECT 1``."""
	if self.pool and self._is_valid:
	return self.pool
	async with self._lock:
	if self.pool and self._is_valid:
	return self.pool

[gemini-code-assist-2]

When a connection-level failure occurs on a direct connection (non-pool), setting self.conn = None without explicitly closing it leaks the underlying database connection and socket. We should explicitly close the connection before discarding the reference.

Suggested change

	except Exception as e:
	# A connection-level failure invalidates the pool / drops the conn.
	if self.conn and self.is_pool:
	self.conn._is_valid = False
	self.conn._last_error = str(e)
	elif self.conn and not self.is_pool:
	self.conn = None
	raise
	except Exception as e:
	# A connection-level failure invalidates the pool / drops the conn.
	if self.conn and self.is_pool:
	self.conn._is_valid = False
	self.conn._last_error = str(e)
	elif self.conn and not self.is_pool:
	try:
	await self.conn.close()
	except Exception:
	pass
	self.conn = None
	raise

[gemini-code-assist-2]

There are two critical transaction management issues in _execute_with_connection:

1. Direct SQL Transaction Control: Executing "COMMIT" and "ROLLBACK" directly via cursor.execute() bypasses psycopg 3's internal transaction state tracking. This can cause psycopg's state to get out of sync, leading to unexpected errors or warnings (especially in autocommit mode). We should use connection.commit() and connection.rollback() instead.
2. Missing Rollback on Error: If force_readonly is False and connection.autocommit is False, psycopg 3 implicitly starts a transaction. If an exception occurs, transaction_started is False, so connection.rollback() is never called. This leaves the connection in a failed transaction state, causing all subsequent queries on that connection to fail with InFailedSqlTransaction.

We should always call await connection.rollback() in the except block, and use connection.commit() / connection.rollback() for transaction control.

    async def _execute_with_connection(
        self, connection: Any, query: Any, params: Any, force_readonly: bool
    ) -> list[RowResult] | None:
        """Execute on a specific connection with read-only + txn handling."""
        transaction_started = False
        try:
            async with connection.cursor(row_factory=dict_row) as cursor:
                if force_readonly:
                    await cursor.execute("BEGIN TRANSACTION READ ONLY")
                    transaction_started = True

                if params:
                    await cursor.execute(query, params)
                else:
                    await cursor.execute(query)

                # For multi-statement input, advance to the last result set.
                while cursor.nextset():
                    pass

                if cursor.description is None:  # no result set (e.g. DDL)
                    if not force_readonly:
                        await connection.commit()
                    elif transaction_started:
                        await connection.rollback()
                        transaction_started = False
                    return None

                rows = await cursor.fetchall()

                if not force_readonly:
                    await connection.commit()
                elif transaction_started:
                    await connection.rollback()
                    transaction_started = False

                return [SqlDriver.RowResult(cells=dict(row)) for row in rows]
        except Exception as e:
            try:
                await connection.rollback()
            except Exception as rollback_error:
                logger.error("Error rolling back transaction: %s", rollback_error)
            logger.error("Error executing query (%s): %s", query, e)
            raise