remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-09-08 01:24:14 +02:00
Code Issues Projects Releases Packages Wiki Activity

Fix for SQL injection issue in table filters, refs #2868

Browse source
This commit is contained in:
Simon Willison 2026-08-06 10:57:29 -07:00
commit c43d89382f
5 changed files with 119 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

56
datasette/filters.py
View file

@ -183,6 +183,14 @@ class Filter:
raise NotImplementedError
def _quote_sqlite_identifier(identifier):
# Preserve the historic always-quoted SQL generated by TemplatedFilter.
escaped = escape_sqlite(identifier)
if escaped == identifier:
return f'"{identifier}"'
return escaped
class TemplatedFilter(Filter):
def __init__(
self,
@ -207,10 +215,14 @@ class TemplatedFilter(Filter):
if self.numeric and converted.isdigit():
converted = int(converted)
if self.no_argument:
kwargs = {"c": column}
kwargs = {"c": _quote_sqlite_identifier(column)}
converted = None
else:
kwargs = {"c": column, "p": f"p{param_counter}", "t": table}
kwargs = {
"c": _quote_sqlite_identifier(column),
"p": f"p{param_counter}",
"t": _quote_sqlite_identifier(table),
}
return self.sql_template.format(**kwargs), converted
def human_clause(self, column, value):
@ -265,49 +277,49 @@ class Filters:
TemplatedFilter(
"exact",
"=",
'"{c}" = :{p}',
"{c} = :{p}",
lambda c, v: "{c} = {v}" if v.isdigit() else '{c} = "{v}"',
),
TemplatedFilter(
"not",
"!=",
'"{c}" != :{p}',
"{c} != :{p}",
lambda c, v: "{c} != {v}" if v.isdigit() else '{c} != "{v}"',
),
TemplatedFilter(
"contains",
"contains",
'"{c}" like :{p}',
"{c} like :{p}",
'{c} contains "{v}"',
format="%{}%",
),
TemplatedFilter(
"endswith",
"ends with",
'"{c}" like :{p}',
"{c} like :{p}",
'{c} ends with "{v}"',
format="%{}",
),
TemplatedFilter(
"startswith",
"starts with",
'"{c}" like :{p}',
"{c} like :{p}",
'{c} starts with "{v}"',
format="{}%",
),
TemplatedFilter("gt", ">", '"{c}" > :{p}', "{c} > {v}", numeric=True),
TemplatedFilter("gt", ">", "{c} > :{p}", "{c} > {v}", numeric=True),
TemplatedFilter(
"gte", "\u2265", '"{c}" >= :{p}', "{c} \u2265 {v}", numeric=True
"gte", "\u2265", "{c} >= :{p}", "{c} \u2265 {v}", numeric=True
),
TemplatedFilter("lt", "<", '"{c}" < :{p}', "{c} < {v}", numeric=True),
TemplatedFilter("lt", "<", "{c} < :{p}", "{c} < {v}", numeric=True),
TemplatedFilter(
"lte", "\u2264", '"{c}" <= :{p}', "{c} \u2264 {v}", numeric=True
"lte", "\u2264", "{c} <= :{p}", "{c} \u2264 {v}", numeric=True
),
TemplatedFilter("like", "like", '"{c}" like :{p}', '{c} like "{v}"'),
TemplatedFilter("like", "like", "{c} like :{p}", '{c} like "{v}"'),
TemplatedFilter(
"notlike", "not like", '"{c}" not like :{p}', '{c} not like "{v}"'
"notlike", "not like", "{c} not like :{p}", '{c} not like "{v}"'
),
TemplatedFilter("glob", "glob", '"{c}" glob :{p}', '{c} glob "{v}"'),
TemplatedFilter("glob", "glob", "{c} glob :{p}", '{c} glob "{v}"'),
InFilter(),
NotInFilter(),
]
@ -316,13 +328,13 @@ class Filters:
TemplatedFilter(
"arraycontains",
"array contains",
""":{p} in (select value from json_each([{t}].[{c}]))""",
""":{p} in (select value from json_each({t}.{c}))""",
'{c} contains "{v}"',
),
TemplatedFilter(
"arraynotcontains",
"array does not contain",
""":{p} not in (select value from json_each([{t}].[{c}]))""",
""":{p} not in (select value from json_each({t}.{c}))""",
'{c} does not contain "{v}"',
),
]
@ -330,30 +342,28 @@ class Filters:
else []
)
+ [
TemplatedFilter("date", "date", "date({c}) = :{p}", '"{c}" is on date {v}'),
TemplatedFilter(
"date", "date", 'date("{c}") = :{p}', '"{c}" is on date {v}'
),
TemplatedFilter(
"isnull", "is null", '"{c}" is null', "{c} is null", no_argument=True
"isnull", "is null", "{c} is null", "{c} is null", no_argument=True
),
TemplatedFilter(
"notnull",
"is not null",
'"{c}" is not null',
"{c} is not null",
"{c} is not null",
no_argument=True,
),
TemplatedFilter(
"isblank",
"is blank",
'("{c}" is null or "{c}" = "")',
"({c} is null or {c} = '')",
"{c} is blank",
no_argument=True,
),
TemplatedFilter(
"notblank",
"is not blank",
'("{c}" is not null and "{c}" != "")',
"({c} is not null and {c} != '')",
"{c} is not blank",
no_argument=True,
),

4
datasette/utils/__init__.py
View file

@ -345,6 +345,10 @@ def escape_css_string(s):
def escape_sqlite(s):
if _boring_keyword_re.match(s) and (s.lower() not in reserved_words):
return s
elif "]" in s:
# SQLite does not support escaping ] inside [bracket] quoting, so fall
# back to double-quote quoting, doubling any embedded double quotes
return '"{}"'.format(s.replace('"', '""'))
else:
return f"[{s}]"

29
tests/test_filters.py
View file

@ -78,6 +78,35 @@ def test_build_where(args, expected_where, expected_params):
assert {f"p{i}": param for i, param in enumerate(expected_params)} == actual_params
@pytest.mark.parametrize(
"key,expected_where",
(
(
'has"quote__exact',
'[has"quote] = :p0',
),
(
'has"quote__isnull',
'[has"quote] is null',
),
(
"has]bracket__arraycontains",
':p0 in (select value from json_each([table]."has]bracket"))',
),
),
)
def test_build_where_escapes_column_names(key, expected_where):
filters = Filters(((key, "value"),))
sql_bits, _ = filters.build_where_clauses("table")
assert sql_bits == [expected_where]
def test_build_where_escapes_table_name():
filters = Filters((("tags__arraycontains", "value"),))
sql_bits, _ = filters.build_where_clauses("items]bracket")
assert sql_bits == [':p0 in (select value from json_each("items]bracket"."tags"))']
@pytest.mark.asyncio
async def test_through_filters_from_request(app_client):
request = Request.fake(

29
tests/test_table_api.py
View file

@ -521,6 +521,35 @@ def test_table_filter_queries_multiple_of_same_type(app_client):
] == response.json["rows"]
@pytest.mark.skipif(not detect_json1(), reason="Requires the SQLite json1 module")
def test_table_filters_quote_identifiers():
with make_app_client(
extra_databases={
"demo.db": """
create table items (
id integer primary key,
"name""quote" text,
"tags]bracket" text
);
insert into items values (1, 'Alice', '["red"]');
"""
},
) as client:
exact_query = urllib.parse.urlencode(
{'name"quote__exact': "Alice", "_shape": "arrays"}
)
exact_response = client.get(f"/demo/items.json?{exact_query}")
assert exact_response.status == 200
assert exact_response.json["rows"] == [[1, "Alice", '["red"]']]
array_query = urllib.parse.urlencode(
{"tags]bracket__arraycontains": "red", "_shape": "arrays"}
)
array_response = client.get(f"/demo/items.json?{array_query}")
assert array_response.status == 200
assert array_response.json["rows"] == [[1, "Alice", '["red"]']]
@pytest.mark.skipif(not detect_json1(), reason="Requires the SQLite json1 module")
def test_table_filter_json_arraycontains(app_client):
response = app_client.get("/fixtures/facetable.json?tags__arraycontains=tag1")

24
tests/test_utils.py
View file

@ -212,6 +212,30 @@ def test_detect_fts(open_quote, close_quote):
assert "Street_Tree_List_fts" == utils.detect_fts(conn, "Street_Tree_List")
@pytest.mark.parametrize(
"identifier,expected",
(
("plain", "plain"),
("select", "[select]"),
("has space", "[has space]"),
("has]bracket", '"has]bracket"'),
('has"quote]', '"has""quote]"'),
),
)
def test_escape_sqlite(identifier, expected):
assert utils.escape_sqlite(identifier) == expected
def test_escape_sqlite_closing_bracket_works_in_query():
conn = utils.sqlite3.connect(":memory:")
table = "has]bracket"
escaped_table = utils.escape_sqlite(table)
conn.execute(f"create table {escaped_table} (id integer)")
conn.execute(f"insert into {escaped_table} values (1)")
assert conn.execute(f"select id from {escaped_table}").fetchall() == [(1,)]
conn.close()
@pytest.mark.parametrize("table", ("regular", "has'single quote"))
def test_detect_fts_different_table_names(table):
sql = """