[sandbox-ext] fix: drop dynamic verify SQL (bandit B608) — static query + Python membership

CI bandit -ll flagged B608 at sandbox.py:264 (f-string ANY(ARRAY[...])
with interpolated feature names). Root-cause fix: verify_step now runs a
static 'SELECT extname FROM pg_extension' and verify_ok checks set
membership against the installed extnames — no interpolation, no string-
built-SQL surface, and a more correct check (membership vs count). The
enable_step CREATE EXTENSION stays (identifiers can't be parameterized;
allowlist-validated upstream, the containment). Tests updated from the
count-based exec_out to the extname-list exec_out.
This commit is contained in:
Renn F
2026-07-13 20:05:45 +02:00
committed by Renzo F
parent 3ae0dad3d2
commit 1f769f6315
2 changed files with 20 additions and 14 deletions
+8 -3
View File
@@ -253,7 +253,9 @@ class _PostgresEngine(SandboxEngine):
def verify_step(self, _password: str, features: list[str]) -> list[str] | None: def verify_step(self, _password: str, features: list[str]) -> list[str] | None:
if not features: if not features:
return None return None
names = ",".join(f"'{f}'" for f in features) # Static query — no interpolation, so no string-built-SQL surface; the
# feature membership check happens in verify_ok against the installed
# extname set. Every requested name is allowlist-validated upstream.
return [ return [
"psql", "psql",
"-U", "-U",
@@ -261,16 +263,19 @@ class _PostgresEngine(SandboxEngine):
"-d", "-d",
"sandbox", "sandbox",
"-tAc", "-tAc",
f"SELECT count(*) FROM pg_extension WHERE extname = ANY(ARRAY[{names}])", "SELECT extname FROM pg_extension",
] ]
def verify_ok(self, features: list[str], stdout: bytes) -> bool: def verify_ok(self, features: list[str], stdout: bytes) -> bool:
if not features: if not features:
return True return True
try: try:
return int(stdout.decode().strip()) == len(features) installed = {
line.strip() for line in stdout.decode().splitlines() if line.strip()
}
except (ValueError, AttributeError): except (ValueError, AttributeError):
return False return False
return set(features).issubset(installed)
def connection( def connection(
self, host: str, password: str, features: tuple[str, ...] = () self, host: str, password: str, features: tuple[str, ...] = ()
+12 -11
View File
@@ -372,9 +372,10 @@ def _exec_calls(runner: _FakeRunner, containish: str | None = None) -> list[list
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_provision_pg_features_runs_enable_then_verify() -> None: async def test_provision_pg_features_runs_enable_then_verify() -> None:
# exec_out = "2\n" satisfies the pg verify (count == len(features)). # exec_out is the installed-extname set the static verify query returns;
# verify_ok checks every requested feature is present in it.
runner = _FakeRunner(run_rc=0, exec_rc=0) runner = _FakeRunner(run_rc=0, exec_rc=0)
runner.exec_out = b"2\n" runner.exec_out = b"vector\npostgis\n"
provisioner = SandboxProvisioner(network=_NETWORK, runner=runner) provisioner = SandboxProvisioner(network=_NETWORK, runner=runner)
info = await provisioner.provision( info = await provisioner.provision(
@@ -383,16 +384,16 @@ async def test_provision_pg_features_runs_enable_then_verify() -> None:
pg = info.services["postgres"] pg = info.services["postgres"]
assert pg.features == ("vector", "postgis") assert pg.features == ("vector", "postgis")
# The enable exec creates both extensions; the verify exec counts them — # The enable exec creates both extensions; the verify exec is a static
# assert each by content rather than a magic count. # SELECT (no interpolation) — verify_ok does the membership check against
# the installed set, so provision succeeding proves both are present.
execs = _exec_calls(runner, "psql") execs = _exec_calls(runner, "psql")
enable = next(c for c in execs if "CREATE EXTENSION" in " ".join(c)) enable = next(c for c in execs if "CREATE EXTENSION" in " ".join(c))
verify = next(c for c in execs if "pg_extension" in " ".join(c)) verify = next(c for c in execs if "pg_extension" in " ".join(c))
assert "CREATE EXTENSION IF NOT EXISTS vector" in " ".join(enable) assert "CREATE EXTENSION IF NOT EXISTS vector" in " ".join(enable)
assert "CREATE EXTENSION IF NOT EXISTS postgis" in " ".join(enable) assert "CREATE EXTENSION IF NOT EXISTS postgis" in " ".join(enable)
assert "ON_ERROR_STOP=1" in enable assert "ON_ERROR_STOP=1" in enable
assert "vector" in " ".join(verify) assert "SELECT extname FROM pg_extension" in " ".join(verify)
assert "postgis" in " ".join(verify)
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -472,7 +473,7 @@ def _run_call(runner: _FakeRunner) -> list[str]:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_provision_pg_features_uses_kitchen_sink_image() -> None: async def test_provision_pg_features_uses_kitchen_sink_image() -> None:
runner = _FakeRunner(run_rc=0, exec_rc=0) runner = _FakeRunner(run_rc=0, exec_rc=0)
runner.exec_out = b"1\n" runner.exec_out = b"vector\n"
provisioner = SandboxProvisioner(network=_NETWORK, runner=runner) provisioner = SandboxProvisioner(network=_NETWORK, runner=runner)
await provisioner.provision( await provisioner.provision(
@@ -582,11 +583,11 @@ async def test_provision_failed_enable_tears_down_and_raises() -> None:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_provision_failed_verify_raises_with_image_hint() -> None: async def test_provision_failed_verify_raises_with_image_hint() -> None:
"""A successful enable but a verify count that's short (the image is missing """A successful enable but a verify that comes up short (the image is missing
the extension files) is fatal with the 'image may be missing' hint.""" the extension files) is fatal with the 'image may be missing' hint."""
# 2 features requested but verify reports only 1 present. # 2 features requested but verify reports only vector installed.
runner = _FakeRunner(run_rc=0, exec_rc=0) runner = _FakeRunner(run_rc=0, exec_rc=0)
runner.exec_out = b"1\n" runner.exec_out = b"vector\n"
provisioner = SandboxProvisioner(network=_NETWORK, runner=runner) provisioner = SandboxProvisioner(network=_NETWORK, runner=runner)
with pytest.raises(SandboxProvisionError, match="missing the extension"): with pytest.raises(SandboxProvisionError, match="missing the extension"):
@@ -598,7 +599,7 @@ async def test_provision_failed_verify_raises_with_image_hint() -> None:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_as_payload_surfaces_available_extensions() -> None: async def test_as_payload_surfaces_available_extensions() -> None:
runner = _FakeRunner(run_rc=0, exec_rc=0) runner = _FakeRunner(run_rc=0, exec_rc=0)
runner.exec_out = b"1\n" runner.exec_out = b"vector\n"
provisioner = SandboxProvisioner(network=_NETWORK, runner=runner) provisioner = SandboxProvisioner(network=_NETWORK, runner=runner)
info = await provisioner.provision( info = await provisioner.provision(