allow_sql config option to disable custom SQL, closes #284

datasette/app.py

@@ -80,6 +80,9 @@ CONFIG_OPTIONS = (
     ConfigOption("suggest_facets", True, """
         Calculate and display suggested facets
     """.strip()),
+    ConfigOption("allow_sql", True, """
+        Allow arbitrary SQL queries via ?sql= parameter
+    """.strip()),
 )
 DEFAULT_CONFIG = {
     option.name: option.default

datasette/templates/database.html

@@ -16,11 +16,13 @@
 
 {% block description_source_license %}{% include "_description_source_license.html" %}{% endblock %}
 
-<form class="sql" action="/{{ database }}-{{ database_hash }}" method="get">
-    <h3>Custom SQL query</h3>
-    <p><textarea name="sql">select * from {{ tables[0].name|escape_sqlite }}</textarea></p>
-    <p><input type="submit" value="Run SQL"></p>
-</form>
+{% if config.allow_sql %}
+    <form class="sql" action="/{{ database }}-{{ database_hash }}" method="get">
+        <h3>Custom SQL query</h3>
+        <p><textarea name="sql">select * from {{ tables[0].name|escape_sqlite }}</textarea></p>
+        <p><input type="submit" value="Run SQL"></p>
+    </form>
+{% endif %}
 
 {% for table in tables %}
 {% if show_hidden or not table.hidden %}

datasette/templates/query.html

@@ -25,7 +25,7 @@
 
 <form class="sql" action="/{{ database }}-{{ database_hash }}{% if canned_query %}/{{ canned_query }}{% endif %}" method="get">
     <h3>Custom SQL query{% if rows %} returning {% if truncated %}more than {% endif %}{{ "{:,}".format(rows|length) }} row{% if rows|length == 1 %}{% else %}s{% endif %}{% endif %}</h3>
-    {% if editable %}
+    {% if editable and config.allow_sql %}
         <p><textarea name="sql">{% if query and query.sql %}{{ query.sql }}{% else %}select * from {{ tables[0].name|escape_sqlite }}{% endif %}</textarea></p>
     {% else %}
         <pre>{% if query %}{{ query.sql }}{% endif %}</pre>

datasette/views/database.py

@@ -11,6 +11,8 @@ class DatabaseView(BaseView):
 
     async def data(self, request, name, hash):
         if request.args.get("sql"):
+            if not self.ds.config["allow_sql"]:
+                raise DatasetteError("sql= is not allowed", status=400)
             sql = request.raw_args.pop("sql")
             validate_sql_select(sql)
             return await self.custom_sql(request, name, hash, sql)

docs/config.rst

@@ -85,3 +85,10 @@ allow_download
 Should users be able to download the original SQLite database using a link on the database index page? This is turned on by default - to disable database downloads, use the following::
 
     datasette mydatabase.db --config allow_download:off
+
+allow_sql
+---------
+
+Enable/disable the ability for users to run custom SQL directly against a database. To disable this feature, run::
+
+    datasette mydatabase.db --config allow_sql:off

tests/test_api.py

@@ -367,6 +367,16 @@ def test_invalid_custom_sql(app_client):
     assert 'Statement must be a SELECT' == response.json['error']
 
 
+def test_allow_sql_off():
+    for client in app_client(config={
+        'allow_sql': False,
+    }):
+        assert 400 == client.get(
+            "/test_tables.json?sql=select+sleep(0.01)",
+            gather_request=False
+        ).status
+
+
 def test_table_json(app_client):
     response = app_client.get('/test_tables/simple_primary_key.json?_shape=objects', gather_request=False)
     assert response.status == 200
@@ -916,7 +926,8 @@ def test_config_json(app_client):
         "sql_time_limit_ms": 200,
         "allow_download": True,
         "allow_facet": True,
-        "suggest_facets": True
+        "suggest_facets": True,
+        "allow_sql": True,
     } == response.json
 
 

tests/test_html.py

@@ -495,6 +495,27 @@ def test_allow_download_off():
         assert 403 == response.status
 
 
+def test_allow_sql_on(app_client):
+    response = app_client.get(
+        "/test_tables",
+        gather_request=False
+    )
+    soup = Soup(response.body, 'html.parser')
+    assert len(soup.findAll('textarea', {'name': 'sql'}))
+
+
+def test_allow_sql_off():
+    for client in app_client(config={
+        'allow_sql': False,
+    }):
+        response = client.get(
+            "/test_tables",
+            gather_request=False
+        )
+        soup = Soup(response.body, 'html.parser')
+        assert not len(soup.findAll('textarea', {'name': 'sql'}))
+
+
 def assert_querystring_equal(expected, actual):
     assert sorted(expected.split('&')) == sorted(actual.split('&'))