Better PRAGMA error message, closes #1185

This commit is contained in:
Simon Willison 2021-01-12 14:26:19 -08:00
commit 640ac7071b
3 changed files with 19 additions and 5 deletions

View file

@ -187,7 +187,9 @@ allowed_pragmas = (
disallawed_sql_res = [
(
re.compile(f"pragma(?!_({'|'.join(allowed_pragmas)}))"),
"Statement may not contain PRAGMA",
"Statement contained a disallowed PRAGMA. Allowed pragma functions are {}".format(
", ".join("pragma_{}()".format(pragma) for pragma in allowed_pragmas)
),
)
]

View file

@ -50,10 +50,7 @@ SQLite string escaping rules will be applied to values passed using named
parameters - they will be wrapped in quotes and their content will be correctly
escaped.
Datasette disallows custom SQL containing the string PRAGMA, as SQLite pragma
statements can be used to change database settings at runtime. If you need to
include the string "pragma" in a query you can do so safely using a named
parameter.
Datasette disallows custom SQL queries containing the string PRAGMA (with a small number `of exceptions <https://github.com/simonw/datasette/issues/761>`__) as SQLite pragma statements can be used to change database settings at runtime. If you need to include the string "pragma" in a query you can do so safely using a named parameter.
.. _sql_views:

View file

@ -1,4 +1,5 @@
from bs4 import BeautifulSoup as Soup
from datasette.utils import allowed_pragmas
from .fixtures import ( # noqa
app_client,
app_client_base_url_prefix,
@ -124,6 +125,20 @@ def test_invalid_custom_sql(app_client):
assert "Statement must be a SELECT" in response.text
def test_disallowed_custom_sql_pragma(app_client):
response = app_client.get(
"/fixtures?sql=SELECT+*+FROM+pragma_not_on_allow_list('idx52')"
)
assert response.status == 400
pragmas = ", ".join("pragma_{}()".format(pragma) for pragma in allowed_pragmas)
assert (
"Statement contained a disallowed PRAGMA. Allowed pragma functions are {}".format(
pragmas
)
in response.text
)
def test_sql_time_limit(app_client_shorter_time_limit):
response = app_client_shorter_time_limit.get("/fixtures?sql=select+sleep(0.5)")
assert 400 == response.status