egutierrez
40400c0b88
CRÍTICO: read_only=True protege la base de datos pero NO el sistema de ficheros. Un SELECT con read_csv/read_blob/glob/COPY...TO podía leer ficheros arbitrarios (claves SSH) o escribirlos (camino a RCE). Añadido parámetro sandbox (default True) que abre la conexión con enable_external_access=false, bloqueando todo acceso a FS/red desde la query. Los SELECT normales sobre tablas siguen funcionando. Único consumidor (osint_db /api/query) queda protegido sin cambios. Tests nuevos: sandbox bloquea read_csv; sandbox=False lo permite. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
"""Tests para duckdb_query_readonly."""
|
|
|
|
import datetime
|
|
import decimal
|
|
|
|
import duckdb
|
|
import pytest
|
|
|
|
from .duckdb_query_readonly import duckdb_query_readonly
|
|
|
|
|
|
@pytest.fixture
|
|
def db(tmp_path):
|
|
"""Crea una base DuckDB temporal con datos de ejemplo y devuelve su path."""
|
|
path = str(tmp_path / "test.duckdb")
|
|
con = duckdb.connect(path)
|
|
con.execute(
|
|
"CREATE TABLE ventas ("
|
|
" id INTEGER,"
|
|
" region VARCHAR,"
|
|
" total DECIMAL(10,2),"
|
|
" fecha DATE"
|
|
")"
|
|
)
|
|
con.execute(
|
|
"INSERT INTO ventas VALUES "
|
|
"(1, 'norte', 120.50, DATE '2026-01-01'), "
|
|
"(2, 'sur', 80.00, DATE '2026-01-02'), "
|
|
"(3, 'norte', 45.25, DATE '2026-01-03')"
|
|
)
|
|
con.close()
|
|
return path
|
|
|
|
|
|
def test_query_ok_devuelve_filas_como_dicts(db):
|
|
res = duckdb_query_readonly(db, "SELECT id, region FROM ventas ORDER BY id")
|
|
assert res["status"] == "ok"
|
|
assert res["columns"] == ["id", "region"]
|
|
assert res["row_count"] == 3
|
|
assert res["truncated"] is False
|
|
assert res["rows"][0] == {"id": 1, "region": "norte"}
|
|
assert res["rows"][1] == {"id": 2, "region": "sur"}
|
|
|
|
|
|
def test_query_con_params_posicionales(db):
|
|
res = duckdb_query_readonly(
|
|
db,
|
|
"SELECT id FROM ventas WHERE region = ? ORDER BY id",
|
|
params=["norte"],
|
|
)
|
|
assert res["status"] == "ok"
|
|
assert res["row_count"] == 2
|
|
assert [row["id"] for row in res["rows"]] == [1, 3]
|
|
|
|
|
|
def test_sql_invalido_devuelve_status_error(db):
|
|
res = duckdb_query_readonly(db, "SELECT * FROM tabla_que_no_existe")
|
|
assert res["status"] == "error"
|
|
assert "error" in res
|
|
assert isinstance(res["error"], str) and res["error"]
|
|
|
|
|
|
def test_db_inexistente_devuelve_status_error(tmp_path):
|
|
missing = str(tmp_path / "no_existe.duckdb")
|
|
res = duckdb_query_readonly(missing, "SELECT 1")
|
|
assert res["status"] == "error"
|
|
assert "error" in res
|
|
|
|
|
|
def test_truncado_a_max_rows(db):
|
|
res = duckdb_query_readonly(db, "SELECT id FROM ventas ORDER BY id", max_rows=2)
|
|
assert res["status"] == "ok"
|
|
assert res["row_count"] == 2
|
|
assert res["truncated"] is True
|
|
assert [row["id"] for row in res["rows"]] == [1, 2]
|
|
|
|
|
|
def test_valores_no_serializables_se_convierten(db):
|
|
res = duckdb_query_readonly(
|
|
db,
|
|
"SELECT total, fecha FROM ventas WHERE id = ?",
|
|
params=[1],
|
|
)
|
|
assert res["status"] == "ok"
|
|
row = res["rows"][0]
|
|
# Decimal -> float
|
|
assert isinstance(row["total"], float)
|
|
assert row["total"] == pytest.approx(120.50)
|
|
# date -> isoformat str
|
|
assert isinstance(row["fecha"], str)
|
|
assert row["fecha"] == "2026-01-01"
|
|
# Verificamos que NO quedan tipos crudos no serializables.
|
|
assert not isinstance(row["total"], decimal.Decimal)
|
|
assert not isinstance(row["fecha"], datetime.date)
|
|
|
|
|
|
def test_sandbox_por_defecto_bloquea_acceso_a_ficheros(db, tmp_path):
|
|
"""Por defecto (sandbox=True) la query no puede tocar el sistema de ficheros."""
|
|
csv = tmp_path / "externo.csv"
|
|
csv.write_text("a\n1\n")
|
|
res = duckdb_query_readonly(db, f"SELECT * FROM read_csv_auto('{csv}')")
|
|
assert res["status"] == "error"
|
|
assert (
|
|
"access" in res["error"].lower() or "permission" in res["error"].lower()
|
|
)
|
|
# El SELECT normal sobre la propia base sigue funcionando con el sandbox.
|
|
ok = duckdb_query_readonly(db, "SELECT count(*) AS n FROM ventas")
|
|
assert ok["status"] == "ok" and ok["rows"][0]["n"] == 3
|
|
|
|
|
|
def test_sandbox_off_permite_leer_csv_del_disco(db, tmp_path):
|
|
"""Con sandbox=False (uso interno confiable) sí puede leer ficheros."""
|
|
csv = tmp_path / "externo.csv"
|
|
csv.write_text("a\n1\n2\n")
|
|
res = duckdb_query_readonly(
|
|
db, f"SELECT count(*) AS n FROM read_csv_auto('{csv}')", sandbox=False
|
|
)
|
|
assert res["status"] == "ok"
|
|
assert res["rows"][0]["n"] == 2
|