diff --git a/datasette/app.py b/datasette/app.py
index 2f89d17c..a7c3c66a 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -110,7 +110,6 @@ CONFIG_OPTIONS = (
         "Allow users to download the original SQLite database files",
     ),
     ConfigOption("suggest_facets", True, "Calculate and display suggested facets"),
-    ConfigOption("allow_sql", True, "Allow arbitrary SQL queries via ?sql= parameter"),
     ConfigOption(
         "default_cache_ttl",
         5,
diff --git a/datasette/default_permissions.py b/datasette/default_permissions.py
index a2f4a315..e750acbf 100644
--- a/datasette/default_permissions.py
+++ b/datasette/default_permissions.py
@@ -34,3 +34,11 @@ def permission_allowed(datasette, actor, action, resource):
         if allow is None:
             return True
         return actor_matches_allow(actor, allow)
+    elif action == "execute-sql":
+        # Use allow_sql block from database block, or from top-level
+        database_allow_sql = datasette.metadata("allow_sql", database=resource)
+        if database_allow_sql is None:
+            database_allow_sql = datasette.metadata("allow_sql")
+        if database_allow_sql is None:
+            return True
+        return actor_matches_allow(actor, database_allow_sql)
diff --git a/datasette/templates/database.html b/datasette/templates/database.html
index 100faee4..5ae51ef7 100644
--- a/datasette/templates/database.html
+++ b/datasette/templates/database.html
@@ -22,7 +22,7 @@
 
 {% block description_source_license %}{% include "_description_source_license.html" %}{% endblock %}
 
-{% if config.allow_sql %}
+{% if allow_execute_sql %}
     <form class="sql" action="{{ database_url(database) }}" method="get">
         <h3>Custom SQL query</h3>
         <p><textarea id="sql-editor" name="sql">{% if tables %}select * from {{ tables[0].name|escape_sqlite }}{% else %}select sqlite_version(){% endif %}</textarea></p>
diff --git a/datasette/templates/query.html b/datasette/templates/query.html
index 7771b101..c65953fb 100644
--- a/datasette/templates/query.html
+++ b/datasette/templates/query.html
@@ -35,7 +35,7 @@
 <form class="sql" action="{{ database_url(database) }}{% if canned_query %}/{{ canned_query }}{% endif %}" method="{% if canned_write %}post{% else %}get{% endif %}">
     <h3>Custom SQL query{% if display_rows %} returning {% if truncated %}more than {% endif %}{{ "{:,}".format(display_rows|length) }} row{% if display_rows|length == 1 %}{% else %}s{% endif %}{% endif %} <span class="show-hide-sql">{% if hide_sql %}(<a href="{{ path_with_removed_args(request, {'_hide_sql': '1'}) }}">show</a>){% else %}(<a href="{{ path_with_added_args(request, {'_hide_sql': '1'}) }}">hide</a>){% endif %}</span></h3>
     {% if not hide_sql %}
-        {% if editable and config.allow_sql %}
+        {% if editable and allow_execute_sql %}
             <p><textarea id="sql-editor" name="sql">{% if query and query.sql %}{{ query.sql }}{% else %}select * from {{ tables[0].name|escape_sqlite }}{% endif %}</textarea></p>
         {% else %}
             <pre id="sql-query">{% if query %}{{ query.sql }}{% endif %}</pre>
diff --git a/datasette/templates/table.html b/datasette/templates/table.html
index 1289e125..373fd576 100644
--- a/datasette/templates/table.html
+++ b/datasette/templates/table.html
@@ -109,7 +109,7 @@
 </div>
 {% endif %}
 
-{% if query.sql and config.allow_sql %}
+{% if query.sql and allow_execute_sql %}
     <p><a class="not-underlined" title="{{ query.sql }}" href="{{ database_url(database) }}?{{ {'sql': query.sql}|urlencode|safe }}{% if query.params %}&amp;{{ query.params|urlencode|safe }}{% endif %}">&#x270e; <span class="underlined">View and edit SQL</span></a></p>
 {% endif %}
 
diff --git a/datasette/views/database.py b/datasette/views/database.py
index e1b29c27..ee99bc2d 100644
--- a/datasette/views/database.py
+++ b/datasette/views/database.py
@@ -26,8 +26,6 @@ class DatabaseView(DataView):
         self.ds.update_with_inherited_metadata(metadata)
 
         if request.args.get("sql"):
-            if not self.ds.config("allow_sql"):
-                raise DatasetteError("sql= is not allowed", status=400)
             sql = request.args.get("sql")
             validate_sql_select(sql)
             return await QueryView(self.ds).data(
@@ -90,6 +88,9 @@ class DatabaseView(DataView):
                 "private": not await self.ds.permission_allowed(
                     None, "view-database", database
                 ),
+                "allow_execute_sql": await self.ds.permission_allowed(
+                    request.actor, "execute-sql", database, default=True
+                ),
             },
             {
                 "show_hidden": request.args.get("_show_hidden"),
@@ -289,6 +290,9 @@ class QueryView(DataView):
                 "columns": columns,
                 "query": {"sql": sql, "params": params},
                 "private": private,
+                "allow_execute_sql": await self.ds.permission_allowed(
+                    request.actor, "execute-sql", database, default=True
+                ),
             },
             extra_template,
             templates,
diff --git a/datasette/views/table.py b/datasette/views/table.py
index 4cec0cda..91245293 100644
--- a/datasette/views/table.py
+++ b/datasette/views/table.py
@@ -342,8 +342,10 @@ class TableView(RowTableShared):
         extra_wheres_for_ui = []
         # Add _where= from querystring
         if "_where" in request.args:
-            if not self.ds.config("allow_sql"):
-                raise DatasetteError("_where= is not allowed", status=400)
+            if not await self.ds.permission_allowed(
+                request.actor, "execute-sql", resource=database, default=True,
+            ):
+                raise DatasetteError("_where= is not allowed", status=403)
             else:
                 where_clauses.extend(request.args.getlist("_where"))
                 extra_wheres_for_ui = [
@@ -839,6 +841,9 @@ class TableView(RowTableShared):
                 "next": next_value and str(next_value) or None,
                 "next_url": next_url,
                 "private": private,
+                "allow_execute_sql": await self.ds.permission_allowed(
+                    request.actor, "execute-sql", database, default=True
+                ),
             },
             extra_template,
             (
diff --git a/docs/authentication.rst b/docs/authentication.rst
index 34d46511..f7281db4 100644
--- a/docs/authentication.rst
+++ b/docs/authentication.rst
@@ -176,7 +176,7 @@ This works for SQL views as well - you can treat them as if they are tables.
 .. warning::
     Restricting access to tables and views in this way will NOT prevent users from querying them using arbitrary SQL queries, `like this <https://latest.datasette.io/fixtures?sql=select+*+from+facetable>`__ for example.
 
-    If you are restricting access to specific tables you should also use the ``"allow_sql"`` block to prevent users from accessing 
+    If you are restricting access to specific tables you should also use the ``"allow_sql"`` block to prevent users from bypassing the limit with their own SQL queries - see :ref:`authentication_permissions_execute_sql`.
 
 .. _authentication_permissions_query:
 
@@ -203,6 +203,37 @@ To limit access to the ``add_name`` canned query in your ``dogs.db`` database to
         }
     }
 
+.. _authentication_permissions_execute_sql:
+
+Controlling the ability to execute arbitrary SQL
+------------------------------------------------
+
+The ``"allow_sql"`` block can be used to control who is allowed to execute arbitrary SQL queries, both using the form on the database page e.g. https://latest.datasette.io/fixtures or by appending a ``?_where=`` parameter to the table page as seen on https://latest.datasette.io/fixtures/facetable?_where=city_id=1.
+
+To enable just the :ref:`root user<authentication_root>` to execute SQL for all databases in your instance, use the following:
+
+.. code-block:: json
+
+    {
+        "allow_sql": {
+            "id": "root"
+        }
+    }
+
+To limit this ability for just one specific database, use this:
+
+.. code-block:: json
+
+    {
+        "databases": {
+            "mydatabase": {
+                "allow_sql": {
+                    "id": "root"
+                }
+            }
+        }
+    }
+
 .. _authentication_actor_matches_allow:
 
 actor_matches_allow()
diff --git a/docs/config.rst b/docs/config.rst
index da93e40a..56b38613 100644
--- a/docs/config.rst
+++ b/docs/config.rst
@@ -150,15 +150,6 @@ Should users be able to download the original SQLite database using a link on th
 
     datasette mydatabase.db --config allow_download:off
 
-.. _config_allow_sql:
-
-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
-
 .. _config_default_cache_ttl:
 
 default_cache_ttl
diff --git a/docs/json_api.rst b/docs/json_api.rst
index 7d37d425..af98eecd 100644
--- a/docs/json_api.rst
+++ b/docs/json_api.rst
@@ -291,7 +291,7 @@ Special table arguments
     though this could potentially result in errors if the wrong syntax is used.
 
 ``?_where=SQL-fragment``
-    If the :ref:`config_allow_sql` config option is enabled, this parameter
+    If the :ref:`permissions_execute_sql` permission is enabled, this parameter
     can be used to pass one or more additional SQL fragments to be used in the
     `WHERE` clause of the SQL used to query the table.
 
diff --git a/docs/pages.rst b/docs/pages.rst
index f220f94d..ce8f5d06 100644
--- a/docs/pages.rst
+++ b/docs/pages.rst
@@ -29,7 +29,7 @@ Database
 ========
 
 Each database has a page listing the tables, views and canned queries
-available for that database. If the :ref:`config_allow_sql` config option is enabled (it's turned on by default) there will also be an interface for executing arbitrary SQL select queries against the data.
+available for that database. If the :ref:`permissions_execute_sql` permission is enabled (it's on by default) there will also be an interface for executing arbitrary SQL select queries against the data.
 
 Examples:
 
diff --git a/docs/sql_queries.rst b/docs/sql_queries.rst
index 5295a2e0..db72deb7 100644
--- a/docs/sql_queries.rst
+++ b/docs/sql_queries.rst
@@ -12,8 +12,8 @@ you like. You can also construct queries using the filter interface on the
 tables page, then click "View and edit SQL" to open that query in the custom
 SQL editor.
 
-Note that this interface is only available if the :ref:`config_allow_sql` option
-has not been disabled.
+Note that this interface is only available if the :ref:`permissions_execute_sql`
+permission is allowed.
 
 Any Datasette SQL query is reflected in the URL of the page, allowing you to
 bookmark them, share them with others and navigate through previous queries
diff --git a/tests/test_api.py b/tests/test_api.py
index 13a98b6a..1a54edec 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -634,13 +634,6 @@ def test_invalid_custom_sql(app_client):
     assert "Statement must be a SELECT" == response.json["error"]
 
 
-def test_allow_sql_off():
-    with make_app_client(config={"allow_sql": False}) as client:
-        response = client.get("/fixtures.json?sql=select+sleep(0.01)")
-        assert 400 == response.status
-        assert "sql= is not allowed" == response.json["error"]
-
-
 def test_table_json(app_client):
     response = app_client.get("/fixtures/simple_primary_key.json?_shape=objects")
     assert response.status == 200
@@ -1137,9 +1130,9 @@ def test_table_filter_extra_where_invalid(app_client):
 
 
 def test_table_filter_extra_where_disabled_if_no_sql_allowed():
-    with make_app_client(config={"allow_sql": False}) as client:
+    with make_app_client(metadata={"allow_sql": {}}) as client:
         response = client.get("/fixtures/facetable.json?_where=neighborhood='Dogpatch'")
-        assert 400 == response.status
+        assert 403 == response.status
         assert "_where= is not allowed" == response.json["error"]
 
 
@@ -1325,7 +1318,6 @@ def test_config_json(app_client):
         "allow_download": True,
         "allow_facet": True,
         "suggest_facets": True,
-        "allow_sql": True,
         "default_cache_ttl": 5,
         "default_cache_ttl_hashed": 365 * 24 * 60 * 60,
         "num_sql_threads": 3,
diff --git a/tests/test_config_dir.py b/tests/test_config_dir.py
index 490b1f1d..b1f6994f 100644
--- a/tests/test_config_dir.py
+++ b/tests/test_config_dir.py
@@ -10,7 +10,6 @@ from datasette import hookimpl
 
 @hookimpl
 def extra_template_vars():
-    print("this is template vars")
     return {
         "from_plugin": "hooray"
     }
@@ -18,7 +17,6 @@ def extra_template_vars():
 METADATA = {"title": "This is from metadata"}
 CONFIG = {
     "default_cache_ttl": 60,
-    "allow_sql": False,
 }
 CSS = """
 body { margin-top: 3em}
@@ -91,7 +89,6 @@ def test_config(config_dir_client):
     response = config_dir_client.get("/-/config.json")
     assert 200 == response.status
     assert 60 == response.json["default_cache_ttl"]
-    assert not response.json["allow_sql"]
 
 
 def test_plugins(config_dir_client):
diff --git a/tests/test_html.py b/tests/test_html.py
index cb0e0c90..e6933dfe 100644
--- a/tests/test_html.py
+++ b/tests/test_html.py
@@ -924,16 +924,8 @@ def test_allow_download_off():
         assert 403 == response.status
 
 
-def test_allow_sql_on(app_client):
-    response = app_client.get("/fixtures")
-    soup = Soup(response.body, "html.parser")
-    assert len(soup.findAll("textarea", {"name": "sql"}))
-    response = app_client.get("/fixtures/sortable")
-    assert b"View and edit SQL" in response.body
-
-
 def test_allow_sql_off():
-    with make_app_client(config={"allow_sql": False}) as client:
+    with make_app_client(metadata={"allow_sql": {}}) as client:
         response = client.get("/fixtures")
         soup = Soup(response.body, "html.parser")
         assert not len(soup.findAll("textarea", {"name": "sql"}))
diff --git a/tests/test_permissions.py b/tests/test_permissions.py
index 90ba1494..d8c98825 100644
--- a/tests/test_permissions.py
+++ b/tests/test_permissions.py
@@ -186,6 +186,35 @@ def test_view_query(allow, expected_anon, expected_auth):
             assert ">fixtures 🔒</h1>" in auth_response.text
 
 
+@pytest.mark.parametrize(
+    "metadata",
+    [
+        {"allow_sql": {"id": "root"}},
+        {"databases": {"fixtures": {"allow_sql": {"id": "root"}}}},
+    ],
+)
+def test_execute_sql(metadata):
+    with make_app_client(metadata=metadata) as client:
+        form_fragment = '<form class="sql" action="/fixtures"'
+
+        # Anonymous users - should not display the form:
+        assert form_fragment not in client.get("/fixtures").text
+        # This should 403:
+        assert 403 == client.get("/fixtures?sql=select+1").status
+        # ?_where= not allowed on tables:
+        assert 403 == client.get("/fixtures/facet_cities?_where=id=3").status
+
+        # But for logged in user all of these should work:
+        cookies = {"ds_actor": client.ds.sign({"id": "root"}, "actor")}
+        response_text = client.get("/fixtures", cookies=cookies).text
+        assert form_fragment in response_text
+        assert 200 == client.get("/fixtures?sql=select+1", cookies=cookies).status
+        assert (
+            200
+            == client.get("/fixtures/facet_cities?_where=id=3", cookies=cookies).status
+        )
+
+
 def test_query_list_respects_view_query():
     with make_app_client(
         metadata={