Publish packages when releases are published

Simplify the solution to 5de0c1724e - avoid contextvar.

.github/workflows/deploy-latest.yml

@@ -14,24 +14,46 @@ jobs:
   deploy:
     runs-on: ubuntu-latest
     steps:
+    - name: Check deployment prerequisites
+      id: deployment-prerequisites
+      env:
+        GCP_SA_KEY: ${{ secrets.GCP_SA_KEY }}
+        LATEST_DATASETTE_SECRET: ${{ secrets.LATEST_DATASETTE_SECRET }}
+      run: |
+        missing=()
+        for variable in GCP_SA_KEY LATEST_DATASETTE_SECRET; do
+          if [[ -z "${!variable:-}" ]]; then
+            missing+=("$variable")
+          fi
+        done
+        if (( ${#missing[@]} )); then
+          echo "::notice::Skipping deployment because required environment variables are missing: ${missing[*]}"
+          echo "available=false" >> "$GITHUB_OUTPUT"
+        else
+          echo "available=true" >> "$GITHUB_OUTPUT"
+        fi
     - name: Check out datasette
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' }}
       uses: actions/checkout@v7
     - name: Set up Python
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' }}
       uses: actions/setup-python@v6
       with:
         python-version: "3.13"
         cache: pip
     - name: Install Python dependencies
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' }}
       run: |
         python -m pip install --upgrade pip
         python -m pip install . --group dev
         python -m pip install sphinx-to-sqlite==0.1a1
     - name: Run tests
-      if: ${{ github.ref == 'refs/heads/main' }}
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' && github.ref == 'refs/heads/main' }}
       run: |
         pytest -n auto -m "not serial"
         pytest -m "serial"
     - name: Build fixtures.db and other files needed to deploy the demo
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' }}
       run: |-
         python tests/fixtures.py \
           fixtures.db \
@@ -40,13 +62,14 @@ jobs:
           plugins \
           --extra-db-filename extra_database.db
     - name: Build docs.db
-      if: ${{ github.ref == 'refs/heads/main' }}
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' && github.ref == 'refs/heads/main' }}
       run: |-
         cd docs
         DISABLE_SPHINX_INLINE_TABS=1 sphinx-build -b xml . _build
         sphinx-to-sqlite ../docs.db _build
         cd ..
     - name: Set up the alternate-route demo
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' }}
       run: |
         echo '
         from datasette import hookimpl
@@ -58,6 +81,7 @@ jobs:
         ' > plugins/alternative_route.py
         cp fixtures.db fixtures2.db
     - name: And the counters writable stored query demo
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' }}
       run: |
         cat > plugins/counters.py <<EOF
         from datasette import hookimpl
@@ -97,12 +121,15 @@ jobs:
     #     cat metadata.json
     - id: auth
       name: Authenticate to Google Cloud
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' }}
       uses: google-github-actions/auth@v3
       with:
         credentials_json: ${{ secrets.GCP_SA_KEY }}
     - name: Set up Cloud SDK
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' }}
       uses: google-github-actions/setup-gcloud@v3
     - name: Deploy to Cloud Run
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' }}
       env:
         LATEST_DATASETTE_SECRET: ${{ secrets.LATEST_DATASETTE_SECRET }}
       run: |-
@@ -122,7 +149,7 @@ jobs:
             --service "datasette-latest$SUFFIX" \
             --secret $LATEST_DATASETTE_SECRET
     - name: Deploy to docs as well (only for main)
-      if: ${{ github.ref == 'refs/heads/main' }}
+      if: ${{ steps.deployment-prerequisites.outputs.available == 'true' && github.ref == 'refs/heads/main' }}
       run: |-
         # Deploy docs.db to a different service
         datasette publish cloudrun docs.db \

.github/workflows/publish.yml

@@ -2,7 +2,7 @@ name: Publish Python Package
 
 on:
   release:
-    types: [created]
+    types: [published]
 
 permissions:
   contents: read

datasette/app.py

@@ -315,7 +315,7 @@ def _permission_cache_key(actor, action, parent, child):
     actor_key = (
         json.dumps(actor, sort_keys=True, default=repr) if actor is not None else None
     )
-    return (actor_key, action, parent, child)
+    return (actor_key, action.name, parent, action.normalize_child(child))
 
 
 async def favicon(request, send):
@@ -1532,15 +1532,28 @@ class Datasette:
         conn.row_factory = sqlite3.Row
         conn.text_factory = lambda x: str(x, "utf-8", "replace")
         if self.sqlite_extensions and database != INTERNAL_DB_NAME:
+            # Extension loading is only enabled for as long as it takes to
+            # load the configured extensions. Leaving it enabled would let
+            # anyone who can execute SQL call load_extension() themselves.
             conn.enable_load_extension(True)
-            for extension in self.sqlite_extensions:
-                # "extension" is either a string path to the extension
-                # or a 2-item tuple that specifies which entrypoint to load.
-                if isinstance(extension, tuple):
-                    path, entrypoint = extension
-                    conn.execute("SELECT load_extension(?, ?)", [path, entrypoint])
-                else:
-                    conn.execute("SELECT load_extension(?)", [extension])
+            try:
+                for extension in self.sqlite_extensions:
+                    # "extension" is either a string path to the extension
+                    # or a 2-item tuple that specifies which entrypoint to load.
+                    if isinstance(extension, tuple):
+                        path, entrypoint = extension
+                        if sys.version_info >= (3, 12):
+                            conn.load_extension(path, entrypoint=entrypoint)
+                        else:
+                            # Connection.load_extension() only gained the
+                            # entrypoint argument in Python 3.12
+                            conn.execute(
+                                "SELECT load_extension(?, ?)", [path, entrypoint]
+                            )
+                    else:
+                        conn.load_extension(extension)
+            finally:
+                conn.enable_load_extension(False)
         if self.setting("cache_size_kb"):
             conn.execute(f"PRAGMA cache_size=-{self.setting('cache_size_kb')}")
         # pylint: disable=no-member
@@ -1733,8 +1746,145 @@ class Datasette:
         sql, params = await build_allowed_resources_sql(
             self, actor, action, parent=parent, include_is_private=include_is_private
         )
+        if action == "view-table":
+            sql, params = await self._apply_derived_table_permissions_to_sql(
+                sql,
+                params,
+                actor=actor,
+                parent=parent,
+                include_is_private=include_is_private,
+            )
         return ResourcesSQL(sql, params)
 
+    async def _allowed_derived_table_source(
+        self, database, source, *, actor, dependencies
+    ):
+        """Check an immediate source, denying sources that are themselves derived."""
+        if any(
+            TableResource.normalize_child(table)
+            == TableResource.normalize_child(source)
+            for table in dependencies
+        ):
+            return False
+        # The source has no dependency in this map. Evaluate its own permission
+        # and prerequisites without starting another dependency check.
+        verdicts = await self._allowed_many(
+            actions=["view-table"],
+            resource=TableResource(database, source),
+            actor=actor,
+            check_derived=False,
+        )
+        return verdicts["view-table"]
+
+    async def _apply_derived_table_permissions_to_sql(
+        self,
+        sql,
+        params,
+        *,
+        actor,
+        parent,
+        include_is_private,
+    ):
+        databases = (
+            [(parent, self.databases[parent])]
+            if parent in self.databases
+            else ([] if parent is not None else list(self.databases.items()))
+        )
+        dependency_maps = dict(
+            zip(
+                (name for name, _ in databases),
+                await asyncio.gather(
+                    *(db.derived_table_dependencies() for _, db in databases)
+                ),
+            )
+        )
+        dependencies = [
+            (database_name, child, source)
+            for database_name, dependency_map in dependency_maps.items()
+            for child, source in dependency_map.items()
+        ]
+        if not dependencies:
+            return sql, params
+
+        sources = sorted(
+            {(database_name, source) for database_name, _, source in dependencies}
+        )
+        actor_verdicts = await asyncio.gather(
+            *(
+                self._allowed_derived_table_source(
+                    database_name,
+                    source,
+                    actor=actor,
+                    dependencies=dependency_maps[database_name],
+                )
+                for database_name, source in sources
+            )
+        )
+        actor_allowed = dict(zip(sources, actor_verdicts))
+
+        anonymous_allowed = {}
+        if include_is_private:
+            anonymous_verdicts = await asyncio.gather(
+                *(
+                    self._allowed_derived_table_source(
+                        database_name,
+                        source,
+                        actor=None,
+                        dependencies=dependency_maps[database_name],
+                    )
+                    for database_name, source in sources
+                )
+            )
+            anonymous_allowed = dict(zip(sources, anonymous_verdicts))
+
+        wrapped_params = dict(params)
+        derived_rows = [
+            [
+                database_name,
+                child,
+                int(actor_allowed[(database_name, source)]),
+                *(
+                    [int(anonymous_allowed[(database_name, source)])]
+                    if include_is_private
+                    else []
+                ),
+            ]
+            for database_name, child, source in dependencies
+        ]
+        derived_param = "_datasette_derived_permissions"
+        while derived_param in wrapped_params:
+            derived_param += "_"
+        wrapped_params[derived_param] = json.dumps(derived_rows)
+
+        derived_columns = "parent, child, source_allowed"
+        select_columns = "allowed.parent, allowed.child, allowed.reason"
+        if include_is_private:
+            derived_columns += ", source_anonymous_allowed"
+            select_columns += (
+                ", CASE WHEN derived.source_anonymous_allowed = 0 "
+                "THEN 1 ELSE allowed.is_private END AS is_private"
+            )
+        wrapped_sql = f"""
+WITH derived_permissions({derived_columns}) AS (
+  SELECT
+    json_extract(value, '$[0]'),
+    json_extract(value, '$[1]'),
+    json_extract(value, '$[2]')
+    {", json_extract(value, '$[3]')" if include_is_private else ""}
+  FROM json_each(:{derived_param})
+),
+allowed AS (
+{sql}
+)
+SELECT {select_columns}
+FROM allowed
+LEFT JOIN derived_permissions AS derived
+  ON allowed.parent = derived.parent AND allowed.child = derived.child COLLATE NOCASE
+WHERE COALESCE(derived.source_allowed, 1) = 1
+ORDER BY allowed.parent, allowed.child
+""".strip()
+        return wrapped_sql, wrapped_params
+
     async def allowed_resources(
         self,
         action: str,
@@ -1937,6 +2087,12 @@ class Datasette:
             )
             # {"edit-schema": True, "drop-table": True, "insert-row": False}
         """
+        return await self._allowed_many(
+            actions=actions, resource=resource, actor=actor, check_derived=True
+        )
+
+    async def _allowed_many(self, *, actions, resource, actor, check_derived):
+        """Evaluate permissions, optionally applying the one-hop source policy."""
         from datasette.permissions import (
             _permission_check_cache,
             _skip_permission_checks,
@@ -1974,7 +2130,7 @@ class Datasette:
         to_check = []
         for name in expanded:
             if cache is not None:
-                key = _permission_cache_key(actor, name, parent, child)
+                key = _permission_cache_key(actor, self.actions[name], parent, child)
                 if key in cache:
                     final[name] = cache[key]
                     continue
@@ -1990,6 +2146,28 @@ class Datasette:
                 child=child,
             )
 
+        if (
+            check_derived
+            and "view-table" in to_check
+            and raw.get("view-table")
+            and isinstance(resource, TableResource)
+            and parent in self.databases
+        ):
+            dependencies = await self.databases[parent].derived_table_dependencies()
+            source = next(
+                (
+                    source
+                    for table, source in dependencies.items()
+                    if TableResource.normalize_child(table)
+                    == TableResource.normalize_child(child)
+                ),
+                None,
+            )
+            if source is not None:
+                raw["view-table"] = await self._allowed_derived_table_source(
+                    parent, source, actor=actor, dependencies=dependencies
+                )
+
         def resolve(name):
             # final verdict = own rules AND verdict of also_requires chain
             if name in final:
@@ -2007,7 +2185,9 @@ class Datasette:
         # Cache the freshly computed checks
         if cache is not None:
             for name in to_check:
-                cache[_permission_cache_key(actor, name, parent, child)] = final[name]
+                cache[
+                    _permission_cache_key(actor, self.actions[name], parent, child)
+                ] = final[name]
 
         # Log every check (including cache hits) for the debug page,
         # dependencies before the actions that required them
@@ -2449,7 +2629,7 @@ class Datasette:
     ):
         data = {"a": actor}
         if expire_after:
-            expires_at = int(time.time()) + (24 * 60 * 60)
+            expires_at = int(time.time()) + expire_after
             data["e"] = baseconv.base62.encode(expires_at)
         response.set_cookie("ds_actor", self.sign(data, "actor"))
 
@@ -2891,6 +3071,50 @@ class DatasetteRouter:
             receive,
             max_post_body_bytes=self.ds.setting("max_post_body_bytes"),
         )
+        match, view = resolve_routes(self.routes, path)
+        is_static = view is favicon or getattr(view, "_datasette_static", False)
+        original_send = send
+
+        async def send(message):
+            if message["type"] == "http.response.start" and not (
+                is_static and message["status"] in (200, 304)
+            ):
+                # Decide privacy after rendering, including for streaming responses
+                # and error handlers. A public primary resource can still include
+                # private labels, actor navigation, or cookie-dependent content.
+                headers = list(message.get("headers", []))
+                personalized = (
+                    request.actor is not None
+                    or "cookie" in request.headers
+                    or "authorization" in request.headers
+                    or any(key.lower() == b"set-cookie" for key, _ in headers)
+                )
+                if personalized:
+                    headers = [
+                        (key, value)
+                        for key, value in headers
+                        if key.lower() != b"cache-control"
+                    ]
+                    headers.append((b"cache-control", b"private, no-store"))
+
+                # Anonymous responses must not be reused for credentialed requests.
+                # Preserve any additional variation specified by views or plugins.
+                vary = [
+                    part.strip()
+                    for key, value in headers
+                    if key.lower() == b"vary"
+                    for part in value.split(b",")
+                    if part.strip()
+                ]
+                if b"*" not in vary:
+                    for name in (b"Cookie", b"Authorization"):
+                        if name.lower() not in {part.lower() for part in vary}:
+                            vary.append(name)
+                headers = [(k, v) for k, v in headers if k.lower() != b"vary"]
+                headers.append((b"vary", b", ".join(vary)))
+                message = dict(message, headers=headers)
+            await original_send(message)
+
         # Populate request_messages if ds_messages cookie is present
         try:
             request._messages = self.ds.unsign(
@@ -2930,8 +3154,7 @@ class DatasetteRouter:
             return await self.handle_401(request, send, token_error)
         scope_modifications["actor"] = actor or default_actor
         scope = dict(scope, **scope_modifications)
-
-        match, view = resolve_routes(self.routes, path)
+        request.scope = scope
 
         if match is None:
             return await self.handle_404(request, send)

datasette/database.py

@@ -29,7 +29,7 @@ from .utils import (
     table_columns,
 )
 from .utils.sql_analysis import SQLAnalysis, analyze_sql_tables
-from .utils.sqlite import sqlite_hidden_table_names
+from .utils.sqlite import sqlite_derived_table_dependencies, sqlite_hidden_table_names
 
 connections = threading.local()
 
@@ -85,6 +85,7 @@ class Database:
         self.cached_hash = None
         self.cached_size = None
         self._cached_table_counts = None
+        self._cached_derived_table_dependencies = None
         self._write_thread = None
         self._write_queue = None
         self._closed = False
@@ -246,17 +247,29 @@ class Database:
         return_all=False,
         returning_limit=EXECUTE_WRITE_RETURNING_LIMIT,
         transaction=True,
+        time_limit_ms=2000,
     ):
         self._check_not_closed()
         if returning_limit < 0:
             raise ValueError("returning_limit must be >= 0")
 
-        def _inner(conn):
+        def execute_sql(conn):
             cursor = conn.execute(sql, params or [])
             return ExecuteWriteResult.from_cursor(
                 cursor, return_all=return_all, returning_limit=returning_limit
             )
 
+        def _inner(conn):
+            try:
+                if time_limit_ms is None:
+                    return execute_sql(conn)
+                with sqlite_timelimit(conn, time_limit_ms):
+                    return execute_sql(conn)
+            except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
+                if e.args == ("interrupted",):
+                    raise QueryInterrupted(e, sql, params)
+                raise
+
         with trace("sql", database=self.name, sql=sql.strip(), params=params):
             results = await self.execute_write_fn(
                 _inner, block=block, request=request, transaction=transaction
@@ -354,6 +367,15 @@ class Database:
                     result = fn(self._write_connection)
             else:
                 result = fn(self._write_connection)
+            if not block:
+                # There is no write thread here, so the write has already
+                # finished. Hand back the same (task_id, reply_future) shape
+                # _send_to_write_thread() returns, with the future already
+                # resolved, so the block=False path below is identical in
+                # both modes.
+                reply_future = asyncio.get_running_loop().create_future()
+                reply_future.set_result(result)
+                result = (uuid.uuid4(), reply_future)
         else:
             result = await self._send_to_write_thread(
                 fn, block=block, transaction=transaction
@@ -425,7 +447,7 @@ class Database:
             )
             self._write_thread.name = f"_execute_writes for database {self.name}"
             self._write_thread.start()
-        task_id = uuid.uuid5(uuid.NAMESPACE_DNS, "datasette.io")
+        task_id = uuid.uuid4()
         loop = asyncio.get_running_loop()
         reply_future = loop.create_future()
         self._write_queue.put(
@@ -759,6 +781,17 @@ class Database:
 
         return hidden_tables
 
+    async def derived_table_dependencies(self):
+        """Return implementation tables and the tables they derive from."""
+        schema_version = (await self.execute("PRAGMA schema_version")).first()[0]
+        if (
+            self._cached_derived_table_dependencies is None
+            or self._cached_derived_table_dependencies[0] != schema_version
+        ):
+            dependencies = await self.execute_fn(sqlite_derived_table_dependencies)
+            self._cached_derived_table_dependencies = (schema_version, dependencies)
+        return self._cached_derived_table_dependencies[1]
+
     async def view_names(self):
         results = await self.execute("select name from sqlite_master where type='view'")
         return [r[0] for r in results.rows]

datasette/default_column_types.py

@@ -6,6 +6,17 @@ import markupsafe
 from datasette import hookimpl
 from datasette.column_types import ColumnType, SQLiteType
 
+_HTTP_URL_RE = re.compile(r"https?://\S+", re.IGNORECASE)
+
+
+def _normalize_http_url(value):
+    if not isinstance(value, str):
+        return None
+    normalized = value.strip()
+    if not _HTTP_URL_RE.fullmatch(normalized):
+        return None
+    return normalized
+
 
 class UrlColumnType(ColumnType):
     name = "url"
@@ -15,7 +26,10 @@ class UrlColumnType(ColumnType):
     async def render_cell(self, value, column, table, database, datasette, request):
         if not value or not isinstance(value, str):
             return None
-        escaped = markupsafe.escape(value.strip())
+        normalized = _normalize_http_url(value)
+        if normalized is None:
+            return markupsafe.escape(value.strip())
+        escaped = markupsafe.escape(normalized)
         return markupsafe.Markup(f'<a href="{escaped}">{escaped}</a>')
 
     async def validate(self, value, datasette):
@@ -23,7 +37,7 @@ class UrlColumnType(ColumnType):
             return None
         if not isinstance(value, str):
             return "URL must be a string"
-        if not re.match(r"^https?://\S+$", value.strip()):
+        if _normalize_http_url(value) is None:
             return "Invalid URL"
         return None
 

datasette/default_permissions/config.py

@@ -92,6 +92,13 @@ class ConfigPermissionProcessor:
             # Tables implicitly reference their parent databases
             self.restricted_databases.update(db for db, _ in self.restricted_tables)
 
+        # Resolve identity keys once per action, rather than scanning the
+        # restriction allowlist for every configured table's allow block.
+        self.restricted_table_keys = {
+            (db, self.action_obj.normalize_child(table) if self.action_obj else table)
+            for db, table in self.restricted_tables
+        }
+
     def evaluate_allow_block(self, allow_block: Any) -> bool | None:
         """Evaluate an allow block against the current actor."""
         if allow_block is None:
@@ -125,8 +132,10 @@ class ConfigPermissionProcessor:
         if parent:
             table_restrictions = (self.restrictions.get("r", {}) or {}).get(parent, {})
             if child:
-                table_actions = table_restrictions.get(child, [])
-                if self.action_checks.intersection(table_actions):
+                child_key = (
+                    self.action_obj.normalize_child(child) if self.action_obj else child
+                )
+                if (parent, child_key) in self.restricted_table_keys:
                     return True
             else:
                 # Parent query should proceed if any child in this database is allowlisted

datasette/default_permissions/restrictions.py

@@ -185,11 +185,15 @@ def restrictions_allow_action(
     # Check table/resource level
     if resource is not None and not isinstance(resource, str) and len(resource) == 2:
         database, table = resource
-        table_allowed = restrictions.get("r", {}).get(database, {}).get(table)
-        if table_allowed is not None:
-            assert isinstance(table_allowed, list)
-            if to_check.intersection(table_allowed):
-                return True
+        action_obj = datasette.actions.get(action)
+        normalize = action_obj.normalize_child if action_obj else lambda name: name
+        for table_name, table_allowed in (
+            restrictions.get("r", {}).get(database, {}).items()
+        ):
+            if normalize(table_name) == normalize(table):
+                assert isinstance(table_allowed, list)
+                if to_check.intersection(table_allowed):
+                    return True
 
     # This action is not explicitly allowed, so reject it
     return False

datasette/default_permissions/sqlite_statistics.py

@@ -0,0 +1,25 @@
+"""Default table-access policy for SQLite optimizer statistics."""
+
+import json
+
+from datasette import hookimpl
+from datasette.permissions import PermissionSQL
+
+
+@hookimpl
+def permission_resources_sql(action):
+    if action != "view-table":
+        return None
+    return PermissionSQL(
+        sql="""
+            SELECT database_name AS parent, value AS child, 0 AS allow,
+                'SQLite statistics tables are denied by default' AS reason
+            FROM catalog_databases
+            CROSS JOIN json_each(:sqlite_statistics_names)
+        """,
+        params={
+            "sqlite_statistics_names": json.dumps(
+                ["sqlite_stat1", "sqlite_stat2", "sqlite_stat3", "sqlite_stat4"]
+            )
+        },
+    )

datasette/filters.py

@@ -2,7 +2,7 @@ import json
 from typing import ClassVar
 
 from datasette import hookimpl
-from datasette.resources import DatabaseResource
+from datasette.resources import DatabaseResource, TableResource
 from datasette.utils.asgi import BadRequest
 from datasette.views.base import DatasetteError
 
@@ -51,13 +51,20 @@ def search_filters(request, database, table, datasette):
         human_descriptions = []
         extra_context = {}
 
-        # Figure out which fts_table to use
+        # Figure out which trusted fts_table to use. Query string parameters can
+        # repeat this mapping (for backwards compatibility), but must not select
+        # a different table or primary key.
         table_metadata = await datasette.table_config(database, table)
         db = datasette.get_database(database)
-        fts_table = request.args.get("_fts_table")
-        fts_table = fts_table or table_metadata.get("fts_table")
+        fts_table = table_metadata.get("fts_table")
         fts_table = fts_table or await db.fts_table(table)
-        fts_pk = request.args.get("_fts_pk", table_metadata.get("fts_pk", "rowid"))
+        fts_pk = table_metadata.get("fts_pk", "rowid")
+        requested_fts_table = request.args.get("_fts_table")
+        requested_fts_pk = request.args.get("_fts_pk")
+        if (requested_fts_table and requested_fts_table != fts_table) or (
+            requested_fts_pk and requested_fts_pk != fts_pk
+        ):
+            raise BadRequest("Invalid _fts_table or _fts_pk")
         search_args = {
             key: request.args[key]
             for key in request.args
@@ -75,6 +82,11 @@ def search_filters(request, database, table, datasette):
         extra_context["supports_search"] = bool(fts_table)
 
         if fts_table and search_args:
+            await datasette.ensure_permission(
+                action="view-table",
+                resource=TableResource(database=database, table=fts_table),
+                actor=request.actor,
+            )
             if "_search" in search_args:
                 # Simple ?_search=xxx
                 search = search_args["_search"]
@@ -135,6 +147,11 @@ def through_filters(request, database, table, datasette):
                 through_table = through_data["table"]
                 other_column = through_data["column"]
                 value = through_data["value"]
+                await datasette.ensure_permission(
+                    action="view-table",
+                    resource=TableResource(database=database, table=through_table),
+                    actor=request.actor,
+                )
                 db = datasette.get_database(database)
                 outgoing_foreign_keys = await db.foreign_keys_for_table(through_table)
                 fk_to_us = next(

datasette/permissions.py

@@ -3,6 +3,10 @@ from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from typing import Any, NamedTuple
 
+_SQLITE_IDENTIFIER_CASE = str.maketrans(
+    "ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz"
+)
+
 # Context variable to track when permission checks should be skipped
 _skip_permission_checks = contextvars.ContextVar(
     "skip_permission_checks", default=False
@@ -49,6 +53,15 @@ class Resource(ABC):
     # Class-level metadata (subclasses must define these)
     name: str = None  # e.g., "table", "database", "model"
     parent_class: type["Resource"] | None = None  # e.g., DatabaseResource for tables
+    case_insensitive_child: bool = False
+
+    @classmethod
+    def normalize_child(cls, child: str | None) -> str | None:
+        """Return a comparison key without changing the resource's display name."""
+        if cls.case_insensitive_child and child is not None:
+            # Match SQLite NOCASE: fold ASCII only, not Unicode lower/casefold.
+            return child.translate(_SQLITE_IDENTIFIER_CASE)
+        return child
 
     # Instance-level optional extra attributes
     reasons: list[str] | None = None
@@ -146,6 +159,11 @@ class Action:
     resource_class: type[Resource] | None = None
     also_requires: str | None = None  # Optional action name that must also be allowed
 
+    def normalize_child(self, child: str | None) -> str | None:
+        if self.resource_class is None:
+            return child
+        return self.resource_class.normalize_child(child)
+
     @property
     def takes_parent(self) -> bool:
         """

datasette/plugins.py

@@ -18,6 +18,7 @@ DEFAULT_PLUGINS = (
     "datasette.actor_auth_cookie",
     "datasette.default_permissions",
     "datasette.default_permissions.tokens",
+    "datasette.default_permissions.sqlite_statistics",
     "datasette.default_actions",
     "datasette.default_column_types",
     "datasette.default_magic_parameters",

datasette/resources.py

@@ -25,6 +25,7 @@ class TableResource(Resource):
 
     name = "table"
     parent_class = DatabaseResource
+    case_insensitive_child = True
 
     def __init__(self, database: str, table: str):
         super().__init__(parent=database, child=table)

datasette/static/column-chooser.js

@@ -472,11 +472,13 @@ class ColumnChooser extends HTMLElement {
           <span class="drag-item-check">
             <input type="checkbox" ${this._checked.has(col) ? "checked" : ""}>
           </span>
-          <span class="drag-item-label">${col}</span>
+          <span class="drag-item-label"></span>
         </label>
         <div class="drop-indicator"></div>
       `;
 
+      li.querySelector(".drag-item-label").textContent = col;
+
       li.querySelector("input").addEventListener("change", (e) => {
         e.target.checked ? this._checked.add(col) : this._checked.delete(col);
         this._updateCounts();

datasette/static/json-format-highlight-1.0.1.js

@@ -1,56 +0,0 @@
-/*
-https://github.com/luyilin/json-format-highlight
-From https://unpkg.com/json-format-highlight@1.0.1/dist/json-format-highlight.js
-MIT Licensed
-*/
-(function (global, factory) {
-  typeof exports === "object" && typeof module !== "undefined"
-    ? (module.exports = factory())
-    : typeof define === "function" && define.amd
-      ? define(factory)
-      : (global.jsonFormatHighlight = factory());
-})(this, function () {
-  "use strict";
-
-  var defaultColors = {
-    keyColor: "dimgray",
-    numberColor: "lightskyblue",
-    stringColor: "lightcoral",
-    trueColor: "lightseagreen",
-    falseColor: "#f66578",
-    nullColor: "cornflowerblue",
-  };
-
-  function index(json, colorOptions) {
-    if (colorOptions === void 0) colorOptions = {};
-
-    if (!json) {
-      return;
-    }
-    if (typeof json !== "string") {
-      json = JSON.stringify(json, null, 2);
-    }
-    var colors = Object.assign({}, defaultColors, colorOptions);
-    json = json.replace(/&/g, "&").replace(/</g, "<").replace(/>/g, ">");
-    return json.replace(
-      /("(\\u[a-zA-Z0-9]{4}|\\[^u]|[^\\"])*"(\s*:)?|\b(true|false|null)\b|-?\d+(?:\.\d*)?(?:[eE][+]?\d+)?)/g,
-      function (match) {
-        var color = colors.numberColor;
-        if (/^"/.test(match)) {
-          color = /:$/.test(match) ? colors.keyColor : colors.stringColor;
-        } else {
-          color = /true/.test(match)
-            ? colors.trueColor
-            : /false/.test(match)
-              ? colors.falseColor
-              : /null/.test(match)
-                ? colors.nullColor
-                : color;
-        }
-        return '<span style="color: ' + color + '">' + match + "</span>";
-      },
-    );
-  }
-
-  return index;
-});

datasette/templates/api_explorer.html

@@ -3,7 +3,6 @@
 {% block title %}API Explorer{% endblock %}
 
 {% block extra_head %}
-<script src="{{ static('json-format-highlight-1.0.1.js') }}"></script>
 {% endblock %}
 
 {% block content %}
@@ -126,7 +125,7 @@ getForm.addEventListener("submit", (ev) => {
     document.getElementById('response-status').textContent = response.status;
     return response.json();
   }).then((data) => {
-    output.querySelector('pre').innerHTML = jsonFormatHighlight(data);
+    output.querySelector('pre').textContent = JSON.stringify(data, null, 2);
     errorList.style.display = 'none';
   }).catch((error) => {
     alert(error);
@@ -174,7 +173,7 @@ postForm.addEventListener("submit", (ev) => {
     } else {
       errorList.style.display = 'none';
     }
-    output.querySelector('pre').innerHTML = jsonFormatHighlight(data);
+    output.querySelector('pre').textContent = JSON.stringify(data, null, 2);
     output.style.display = 'block';
   }).catch(err => {
     alert("Error: " + err);

datasette/templates/debug_allowed.html

@@ -3,7 +3,6 @@
 {% block title %}Allowed Resources{% endblock %}
 
 {% block extra_head %}
-<script src="{{ static('json-format-highlight-1.0.1.js') }}"></script>
 {% include "_permission_ui_styles.html" %}
 {% include "_debug_common_functions.html" %}
 {% endblock %}
@@ -198,7 +197,7 @@ function displayResults(data) {
     }
 
     // Update raw JSON
-    document.getElementById('raw-json').innerHTML = jsonFormatHighlight(data);
+    document.getElementById('raw-json').textContent = JSON.stringify(data, null, 2);
 }
 
 function displayError(data) {
@@ -208,7 +207,7 @@ function displayError(data) {
 
     resultsContent.innerHTML = `<div class="error-message">Error: ${escapeHtml(data.error || 'Unknown error')}</div>`;
 
-    document.getElementById('raw-json').innerHTML = jsonFormatHighlight(data);
+    document.getElementById('raw-json').textContent = JSON.stringify(data, null, 2);
 }
 
 // Disable child input if parent is empty

datasette/templates/debug_check.html

@@ -3,7 +3,6 @@
 {% block title %}Explain a permission decision{% endblock %}
 
 {% block extra_head %}
-<script src="{{ static('json-format-highlight-1.0.1.js') }}"></script>
 {% include "_permission_ui_styles.html" %}
 {% include "_debug_common_functions.html" %}
 <style>
@@ -238,7 +237,7 @@ function displayResult(data) {
     displayRules(data.explanation);
     displayRestrictions(data.explanation.restrictions);
     displayRequirements(data.explanation.required_actions);
-    document.getElementById('raw-json').innerHTML = jsonFormatHighlight(data);
+    document.getElementById('raw-json').textContent = JSON.stringify(data, null, 2);
 }
 
 function displayRules(explanation) {
@@ -298,7 +297,7 @@ function displayError(data) {
     document.getElementById('matching-rules').innerHTML = '';
     document.getElementById('restrictions-section').style.display = 'none';
     document.getElementById('requirements-section').style.display = 'none';
-    document.getElementById('raw-json').innerHTML = jsonFormatHighlight(data);
+    document.getElementById('raw-json').textContent = JSON.stringify(data, null, 2);
 }
 
 form.addEventListener('submit', event => {

datasette/templates/debug_rules.html

@@ -3,7 +3,6 @@
 {% block title %}Permission Rules{% endblock %}
 
 {% block extra_head %}
-<script src="{{ static('json-format-highlight-1.0.1.js') }}"></script>
 {% include "_permission_ui_styles.html" %}
 {% include "_debug_common_functions.html" %}
 {% endblock %}
@@ -185,7 +184,7 @@ function displayResults(data) {
     }
 
     // Update raw JSON
-    document.getElementById('raw-json').innerHTML = jsonFormatHighlight(data);
+    document.getElementById('raw-json').textContent = JSON.stringify(data, null, 2);
 }
 
 function displayError(data) {
@@ -195,7 +194,7 @@ function displayError(data) {
 
     resultsContent.innerHTML = `<div class="error-message">Error: ${escapeHtml(data.error || 'Unknown error')}</div>`;
 
-    document.getElementById('raw-json').innerHTML = jsonFormatHighlight(data);
+    document.getElementById('raw-json').textContent = JSON.stringify(data, null, 2);
 }
 
 </script>

datasette/utils/__init__.py

@@ -820,7 +820,8 @@ def detect_spatialite(conn):
 
 def detect_fts(conn, table):
     """Detect if table has a corresponding FTS virtual table and return it"""
-    rows = conn.execute(detect_fts_sql(table)).fetchall()
+    sql, params = detect_fts_sql(table)
+    rows = conn.execute(sql, params).fetchall()
     if len(rows) == 0:
         return None
     else:
@@ -828,18 +829,26 @@ def detect_fts(conn, table):
 
 
 def detect_fts_sql(table):
-    return r"""
-        select name from sqlite_master
-            where rootpage = 0
-            and (
-                sql like '%VIRTUAL TABLE%USING FTS%content="{table}"%'
-                or sql like '%VIRTUAL TABLE%USING FTS%content=[{table}]%'
-                or (
-                    tbl_name = "{table}"
-                    and sql like '%VIRTUAL TABLE%USING FTS%'
+    escaped_table = table.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+    return (
+        r"""
+            select name from sqlite_master
+                where rootpage = 0
+                and (
+                    sql like :fts_double_quoted escape char(92)
+                    or sql like :fts_bracket_quoted escape char(92)
+                    or (
+                        tbl_name = :table
+                        and sql like '%VIRTUAL TABLE%USING FTS%'
+                    )
                 )
-            )
-    """.format(table=table.replace("'", "''"))
+        """,
+        {
+            "fts_double_quoted": f'%VIRTUAL TABLE%USING FTS%content="{escaped_table}"%',
+            "fts_bracket_quoted": f"%VIRTUAL TABLE%USING FTS%content=[{escaped_table}]%",
+            "table": table,
+        },
+    )
 
 
 def detect_json1(conn=None):
@@ -1557,7 +1566,13 @@ async def row_sql_params_pks(db, table, pk_values):
     if use_rowid:
         select = "rowid, *"
         pks = ["rowid"]
-    wheres = [f'"{pk}"=:p{i}' for i, pk in enumerate(pks)]
+    wheres = []
+    for i, pk in enumerate(pks):
+        escaped_pk = escape_sqlite(pk)
+        # Preserve the historic always-quoted SQL exposed by _extra=query
+        if escaped_pk == pk:
+            escaped_pk = f'"{pk}"'
+        wheres.append(f"{escaped_pk}=:p{i}")
     sql = f"select {select} from {escape_sqlite(table)} where {' AND '.join(wheres)}"
     params = {}
     for i, pk_value in enumerate(pk_values):
@@ -1729,7 +1744,7 @@ def redact_keys(original: dict, key_patterns: Iterable) -> dict:
             return {
                 k: (
                     redact(v)
-                    if not any(pattern in k for pattern in key_patterns)
+                    if not any(pattern in k.casefold() for pattern in key_patterns)
                     else "***"
                 )
                 for k, v in data.items()

datasette/utils/actions_sql.py

@@ -29,6 +29,15 @@ from datasette.utils.permissions import gather_permission_sql_from_hooks
 
 if TYPE_CHECKING:
     from datasette.app import Datasette
+    from datasette.permissions import Action
+
+
+def _child_collation(action: "Action") -> str:
+    """Match resource identity without changing the spelling returned by SQL."""
+    resource_class = action.resource_class
+    if resource_class is not None and resource_class.case_insensitive_child:
+        return "NOCASE"
+    return "BINARY"
 
 
 async def build_allowed_resources_sql(
@@ -149,6 +158,7 @@ async def _build_single_action_sql(
         raise ValueError(f"Unknown action: {action}")
 
     # Get base resources SQL from the resource class
+    child_collation = _child_collation(action_obj)
     base_resources_sql = await action_obj.resource_class.resources_sql(
         datasette, actor=actor
     )
@@ -185,7 +195,7 @@ async def _build_single_action_sql(
         if permission_sql.sql is None:
             continue
         rule_sqls.append(f"""
-            SELECT parent, child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (
+            SELECT parent, child COLLATE {child_collation} AS child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (
                 {permission_sql.sql}
             )
             """.strip())
@@ -299,9 +309,9 @@ async def _build_single_action_sql(
         query_parts.extend(
             ["anon_child_agg AS ("]
             + _anon_agg(
-                "parent, child,",
+                f"parent, child COLLATE {child_collation} AS child,",
                 "parent IS NOT NULL AND child IS NOT NULL",
-                "parent, child",
+                f"parent, child COLLATE {child_collation}",
             )
             + ["),", "anon_parent_agg AS ("]
             + _anon_agg("parent,", "parent IS NOT NULL AND child IS NULL", "parent")
@@ -382,7 +392,8 @@ async def _build_single_action_sql(
         # Wrap each restriction_sql in a subquery to avoid operator precedence issues
         # with UNION ALL inside the restriction SQL statements
         restriction_intersect = "\nINTERSECT\n".join(
-            f"SELECT * FROM ({sql})" for sql in restriction_sqls
+            f"SELECT parent, child COLLATE {child_collation} AS child FROM ({sql})"
+            for sql in restriction_sqls
         )
         # Decompose by NULL-pattern so the final filter can use pure-equality
         # EXISTS lookups (satisfiable via automatic indexes) instead of a
@@ -480,6 +491,7 @@ async def build_permission_rules_sql(
     union_parts = []
     all_params = {}
     restriction_sqls = []
+    child_collation = _child_collation(action_obj)
 
     for permission_sql in permission_sqls:
         all_params.update(permission_sql.params or {})
@@ -493,7 +505,7 @@ async def build_permission_rules_sql(
             continue
 
         union_parts.append(f"""
-            SELECT parent, child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (
+            SELECT parent, child COLLATE {child_collation} AS child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (
                 {permission_sql.sql}
             )
             """.strip())
@@ -564,6 +576,7 @@ async def check_permissions_for_actions(
     verdicts = {}
 
     for i, (action, permission_sqls) in enumerate(zip(unique_actions, gathered)):
+        child_collation = _child_collation(datasette.actions[action])
         prefix = f"a{i}_"
         rule_parts = []
         restriction_parts = []
@@ -589,7 +602,7 @@ async def check_permissions_for_actions(
             if sql is None:
                 continue
             rule_parts.append(
-                f"SELECT parent, child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (\n{sql}\n)"
+                f"SELECT parent, child COLLATE {child_collation} AS child, allow, reason, '{permission_sql.source}' AS source_plugin FROM (\n{sql}\n)"
             )
 
         if not rule_parts:
@@ -623,7 +636,8 @@ async def check_permissions_for_actions(
         if restriction_parts:
             # Database-level restrictions (parent, NULL) match all children
             restriction_intersect = "\nINTERSECT\n".join(
-                f"SELECT * FROM ({sql})" for sql in restriction_parts
+                f"SELECT parent, child COLLATE {child_collation} AS child FROM ({sql})"
+                for sql in restriction_parts
             )
             ctes.append(f"a{i}_restriction AS (\n{restriction_intersect}\n)")
             verdict_sql = f"""({verdict_sql}) AND EXISTS (
@@ -770,6 +784,7 @@ async def _explain_single_action(
     db = datasette.get_internal_database()
     matched_rules = []
     restrictions = []
+    child_collation = _child_collation(datasette.actions[action])
 
     for permission_sql in permission_sqls:
         params = dict(permission_sql.params or {})
@@ -784,7 +799,7 @@ async def _explain_single_action(
                 SELECT parent, child, allow, reason
                 FROM ({permission_sql.sql}) AS permission_rules
                 WHERE (parent IS NULL OR parent = :{parent_param})
-                  AND (child IS NULL OR child = :{child_param})
+                  AND (child IS NULL OR child COLLATE {child_collation} = :{child_param})
                 """,
                 params,
             )
@@ -811,7 +826,7 @@ async def _explain_single_action(
                     SELECT EXISTS(
                         SELECT 1 FROM ({permission_sql.restriction_sql}) AS restriction_rules
                         WHERE (parent IS NULL OR parent = :{parent_param})
-                          AND (child IS NULL OR child = :{child_param})
+                          AND (child IS NULL OR child COLLATE {child_collation} = :{child_param})
                     ) AS resource_is_in_allowlist
                     """,
                     params,

datasette/utils/asgi.py

@@ -498,6 +498,8 @@ def asgi_static(root_path, chunk_size=4096, headers=None, content_type=None):
             await asgi_send_html(send, "404: File not found", 404)
             return
 
+    # Only the actual static-file handler can bypass dynamic response privacy.
+    inner_static._datasette_static = True
     return inner_static
 
 

datasette/utils/sql_analysis.py

@@ -1,6 +1,8 @@
+import sys
 from dataclasses import dataclass
 from typing import Literal
 
+from datasette.utils import escape_sqlite
 from datasette.utils.sqlite import SQLiteTableType, sqlite3, sqlite_table_type
 
 SQLOperation = Literal[
@@ -195,6 +197,16 @@ def _allow_authorizer_action(*args):
     return sqlite3.SQLITE_OK
 
 
+def _disable_authorizer(conn):
+    # Python 3.11 added support for unregistering an authorizer using None.
+    # On Python 3.10, None is installed as the callback instead, and the next
+    # statement fails with "not authorized" when sqlite3 tries to call it.
+    if sys.version_info >= (3, 11):
+        conn.set_authorizer(None)
+    else:
+        conn.set_authorizer(_allow_authorizer_action)
+
+
 def analyze_sql_tables(
     conn,
     sql: str,
@@ -208,7 +220,9 @@ def analyze_sql_tables(
 
     This function is synchronous and connection-based. It temporarily installs a
     SQLite authorizer, prepares ``EXPLAIN <sql>``, and returns the operation
-    callbacks observed while SQLite compiles the statement.
+    callbacks observed while SQLite compiles the statement. ``CREATE VIEW`` is
+    additionally executed inside a rolled-back savepoint so its source-table reads
+    can be discovered by analyzing a query against the temporary view.
     """
     operations: dict[OperationKey, set[str]] = {}
 
@@ -481,7 +495,7 @@ def analyze_sql_tables(
                         conn, key.table, schema=key.sqlite_schema
                     )
     finally:
-        conn.set_authorizer(None)
+        _disable_authorizer(conn)
 
     has_schema_operation = any(
         key.target_type in {"table", "index", "view", "trigger", "virtual-table"}
@@ -532,7 +546,7 @@ def analyze_sql_tables(
             return None
         return table_kind_cache[(key.sqlite_schema, key.table)]
 
-    return SQLAnalysis(
+    analysis = SQLAnalysis(
         operations=tuple(
             Operation(
                 operation=key.operation,
@@ -549,3 +563,58 @@ def analyze_sql_tables(
             for key, columns in operations.items()
         )
     )
+
+    # SQLite does not resolve the SELECT body of a view when preparing CREATE
+    # VIEW, so its authorizer does not report reads from the view's source
+    # tables. Temporarily create the view, analyze a query against it (which
+    # does resolve the body), then roll the schema change back. Database-level
+    # callers use an isolated writable connection for this analysis.
+    create_view_operations = tuple(
+        operation
+        for operation in analysis.operations
+        if operation.operation == "create" and operation.target_type == "view"
+    )
+    if not create_view_operations:
+        return analysis
+
+    savepoint = "datasette_analyze_create_view"
+    conn.execute(f"SAVEPOINT {savepoint}")
+    try:
+        conn.execute(sql, params if params is not None else {})
+        dependency_reads = []
+        for view_operation in create_view_operations:
+            if view_operation.sqlite_schema is None or view_operation.table is None:
+                raise sqlite3.OperationalError(
+                    "Could not determine the created view name"
+                )
+            quoted_schema = escape_sqlite(view_operation.sqlite_schema)
+            quoted_view = escape_sqlite(view_operation.table)
+            qualified_view = f"{quoted_schema}.{quoted_view}"
+            view_analysis = analyze_sql_tables(
+                conn,
+                f"SELECT * FROM {qualified_view}",
+                database_name=database_name,
+                schema_to_database=schema_to_database,
+            )
+            dependency_reads.extend(
+                operation
+                for operation in view_analysis.operations
+                if operation.operation == "read"
+                and not (
+                    operation.sqlite_schema == view_operation.sqlite_schema
+                    and operation.table == view_operation.table
+                )
+            )
+    finally:
+        conn.execute(f"ROLLBACK TO {savepoint}")
+        conn.execute(f"RELEASE {savepoint}")
+
+    existing_operations = set(analysis.operations)
+    return SQLAnalysis(
+        operations=analysis.operations
+        + tuple(
+            operation
+            for operation in dependency_reads
+            if operation not in existing_operations
+        )
+    )

datasette/utils/sqlite.py

@@ -15,8 +15,17 @@ if hasattr(sqlite3, "enable_callback_tracebacks"):
 _cached_sqlite_version = None
 _cached_supports_returning = None
 SQLiteTableType = Literal["table", "view", "virtual", "shadow"]
+_SQLITE_IDENTIFIER_RE = (
+    r"""(?:"(?:[^"]|"")*"|'(?:[^']|'')*'|`(?:[^`]|``)*`|\[[^\]]*\]|[^\s.()'"`\[\]]+)"""
+)
 _VIRTUAL_TABLE_MODULE_RE = re.compile(
-    r"\bCREATE\s+VIRTUAL\s+TABLE\b.*?\bUSING\s+([^\s(]+)",
+    r"^\s*CREATE\s+VIRTUAL\s+TABLE\b\s*(?:IF\s+NOT\s+EXISTS\s+)?"
+    + _SQLITE_IDENTIFIER_RE
+    + r"(?:\s*\.\s*"
+    + _SQLITE_IDENTIFIER_RE
+    + r")?\s*\bUSING\b\s*("
+    + _SQLITE_IDENTIFIER_RE
+    + r")",
     re.IGNORECASE | re.DOTALL,
 )
 _VIRTUAL_TABLE_SHADOW_SUFFIXES = {
@@ -83,19 +92,53 @@ def sqlite_table_type(
 ) -> SQLiteTableType | None:
     if supports_table_list():
         try:
-            query = "select type from pragma_table_list where name = ?"
-            params: tuple[str, ...] = (table,)
+            # Use the "PRAGMA table_list" statement form rather than the
+            # pragma_table_list(...) table-valued function. The
+            # table-valued function is resolved like an ordinary relation
+            # name, so an attacker-created table or view literally named
+            # "pragma_table_list" can shadow it and spoof the reported
+            # type (e.g. claiming a virtual table is an ordinary table).
+            # The PRAGMA statement form is a distinct piece of SQL syntax
+            # that always invokes SQLite's built-in pragma, so it cannot
+            # be shadowed by a user-created relation.
             if schema is not None:
-                query += " and schema = ?"
-                params = (table, schema)
-            row = conn.execute(query, params).fetchone()
-            if row is not None and row[0] in {"table", "view", "virtual", "shadow"}:
-                return row[0]
+                query = f"PRAGMA {_quote_identifier(schema)}.table_list"
+            else:
+                query = "PRAGMA table_list"
+            cursor = conn.execute(query)
+            columns = [description[0] for description in cursor.description]
+            for row in cursor.fetchall():
+                record = dict(zip(columns, row))
+                if record.get("name") != table:
+                    continue
+                if schema is not None and record.get("schema") != schema:
+                    continue
+                row_type = record.get("type")
+                if row_type in {"table", "view", "virtual", "shadow"}:
+                    return row_type
         except sqlite3.DatabaseError:
             pass
     return _sqlite_table_type_from_schema(conn, table, schema=schema)
 
 
+def check_structured_write_table(conn, table: str, *, allow_missing=False):
+    """Validate a row-write target on the connection that will perform the write."""
+    # SQLite resolves identifiers case-insensitively. The create API must not
+    # treat a differently cased existing name as a missing table.
+    row = conn.execute(
+        "select name from main.sqlite_master where name = ? collate nocase "
+        "and type in ('table', 'view')",
+        (table,),
+    ).fetchone()
+    if row is None and allow_missing:
+        return
+    if row is not None and sqlite_table_type(conn, row[0]) == "table":
+        return
+    # Virtual table modules can interpret row writes as administrative operations.
+    # Their shadow tables are internal storage, not independently writable data.
+    raise ValueError("Structured writes require an ordinary table")
+
+
 def sqlite_hidden_table_names(conn, *, schema: str | None = "main") -> list[str]:
     schema_table = _sqlite_schema_table(schema)
     try:
@@ -118,6 +161,63 @@ def sqlite_hidden_table_names(conn, *, schema: str | None = "main") -> list[str]
     return sorted(hidden_tables) + content_fts_tables
 
 
+def sqlite_derived_table_dependencies(
+    conn, *, schema: str | None = "main"
+) -> dict[str, str]:
+    """Return implementation table -> logical/content table dependencies.
+
+    ``PRAGMA table_list`` safely identifies virtual and shadow tables, but
+    does not report which virtual table owns a shadow table or which table is
+    named by an FTS ``content=`` option. Derive those relationships from
+    ``sqlite_master`` DDL and the documented shadow-table suffixes.
+
+    Database errors propagate: failed discovery must not be mistaken for an
+    empty dependency map and cached as permission to skip inheritance.
+    """
+    schema_table = _sqlite_schema_table(schema)
+    rows = conn.execute(
+        f"select name, sql from {schema_table} where type = 'table'"
+    ).fetchall()
+
+    table_names = {row[0] for row in rows}
+    # SQLite identifiers fold ASCII letters only.
+    identifier_case = str.maketrans(
+        "ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz"
+    )
+    canonical_names = {name.translate(identifier_case): name for name in table_names}
+    dependencies = {}
+    for virtual_table, sql in rows:
+        module = _virtual_table_module(sql)
+        if module is None:
+            continue
+
+        # SQLite's documented shadow tables are implementation details of
+        # their logical virtual table.
+        for suffix in _VIRTUAL_TABLE_SHADOW_SUFFIXES.get(module, ()):
+            shadow_table = virtual_table + suffix
+            if shadow_table in table_names:
+                dependencies[shadow_table] = virtual_table
+
+        # An external-content FTS table can expose values fetched from its
+        # content table, so it must also depend on that table's permission.
+        if module in {"fts3", "fts4", "fts5"}:
+            content_table = _fts_external_content_table(sql)
+            if content_table:
+                dependencies[virtual_table] = content_table
+
+        if module in {"fts5vocab", "fts4aux"}:
+            source = _fts_vocabulary_source(sql, module, schema or "main")
+            source = (
+                canonical_names.get(source.translate(identifier_case))
+                if source
+                else None
+            )
+            # An unresolved source is itself derived, so the one-hop policy denies it.
+            dependencies[virtual_table] = source or virtual_table
+
+    return dependencies
+
+
 def _sqlite_table_type_from_schema(
     conn,
     table: str,
@@ -184,10 +284,151 @@ def _quote_identifier(value: str) -> str:
 def _virtual_table_module(sql: str | None) -> str | None:
     if not sql:
         return None
+    match = _VIRTUAL_TABLE_MODULE_RE.search(_strip_sql_comments(sql))
+    if match is None:
+        return None
+    return _unquote_sql_value(match.group(1)).lower()
+
+
+def _fts_external_content_table(sql: str | None) -> str | None:
+    """Extract the external ``content=`` table from an FTS declaration."""
+    if not sql:
+        return None
+    sql = _strip_sql_comments(sql)
     match = _VIRTUAL_TABLE_MODULE_RE.search(sql)
     if match is None:
         return None
-    return match.group(1).strip("\"'[]`").lower()
+    open_paren = sql.find("(", match.end())
+    if open_paren == -1:
+        return None
+    close_paren = sql.rfind(")")
+    if close_paren <= open_paren:
+        return None
+
+    for argument in _split_sql_arguments(sql[open_paren + 1 : close_paren]):
+        key, separator, value = argument.partition("=")
+        if not separator or key.strip().lower() != "content":
+            continue
+        return _unquote_sql_value(value.strip())
+    return None
+
+
+def _fts_vocabulary_source(sql: str, module: str, schema: str) -> str | None:
+    """Resolve a vocabulary source within the current SQLite schema.
+
+    Cross-schema sources cannot be represented by the dependency map and
+    are conservatively left unresolved.
+    """
+    sql = _strip_sql_comments(sql)
+    match = _VIRTUAL_TABLE_MODULE_RE.search(sql)
+    if match is None:
+        return None
+    start = sql.find("(", match.end())
+    end = sql.rfind(")")
+    if start < 0 or end <= start:
+        return None
+    arguments = [
+        _unquote_sql_value(arg.strip())
+        for arg in _split_sql_arguments(sql[start + 1 : end])
+    ]
+    expected = 2 if module == "fts5vocab" else 1
+    if len(arguments) == expected:
+        return arguments[0]
+    if len(arguments) == expected + 1 and arguments[0].lower() == schema.lower():
+        return arguments[1]
+    return None
+
+
+def _split_sql_arguments(arguments: str) -> list[str]:
+    """Split comma-separated SQLite arguments without splitting quoted text."""
+    parts = []
+    start = 0
+    quote = None
+    closing_quote = None
+    index = 0
+    while index < len(arguments):
+        char = arguments[index]
+        if quote is None:
+            if char in {"'", '"', "`", "["}:
+                quote = char
+                closing_quote = "]" if char == "[" else char
+            elif char == ",":
+                parts.append(arguments[start:index])
+                start = index + 1
+        elif char == closing_quote:
+            # Single/double/backtick quoting escapes the delimiter by
+            # doubling it. Square-bracket identifiers do not.
+            if (
+                quote != "["
+                and index + 1 < len(arguments)
+                and arguments[index + 1] == closing_quote
+            ):
+                index += 1
+            else:
+                quote = None
+                closing_quote = None
+        index += 1
+    parts.append(arguments[start:])
+    return parts
+
+
+def _strip_sql_comments(sql: str) -> str:
+    """Remove SQLite comments while preserving quoted strings/identifiers."""
+    output = []
+    quote = None
+    closing_quote = None
+    index = 0
+    while index < len(sql):
+        char = sql[index]
+        next_char = sql[index + 1] if index + 1 < len(sql) else ""
+        if quote is None:
+            if char in {"'", '"', "`", "["}:
+                quote = char
+                closing_quote = "]" if char == "[" else char
+                output.append(char)
+            elif char == "-" and next_char == "-":
+                index += 2
+                while index < len(sql) and sql[index] not in "\r\n":
+                    index += 1
+                output.append(" ")
+                continue
+            elif char == "/" and next_char == "*":
+                index += 2
+                while index + 1 < len(sql) and sql[index : index + 2] != "*/":
+                    index += 1
+                index = min(index + 2, len(sql))
+                output.append(" ")
+                continue
+            else:
+                output.append(char)
+        else:
+            output.append(char)
+            if char == closing_quote:
+                if (
+                    quote != "["
+                    and index + 1 < len(sql)
+                    and sql[index + 1] == closing_quote
+                ):
+                    output.append(sql[index + 1])
+                    index += 1
+                else:
+                    quote = None
+                    closing_quote = None
+        index += 1
+    return "".join(output)
+
+
+def _unquote_sql_value(value: str) -> str:
+    if len(value) < 2:
+        return value
+    pairs = {"'": "'", '"': '"', "`": "`", "[": "]"}
+    closing = pairs.get(value[0])
+    if closing is None or value[-1] != closing:
+        return value
+    unquoted = value[1:-1]
+    if value[0] != "[":
+        unquoted = unquoted.replace(closing * 2, closing)
+    return unquoted
 
 
 def _is_fts_content_virtual_table(sql: str | None) -> bool:

datasette/version.py

@@ -1,2 +1,2 @@
-__version__ = "1.0a38"
+__version__ = "1.0a39"
 __version_info__ = tuple(__version__.split("."))

datasette/views/database.py

@@ -40,7 +40,11 @@ from datasette.write_sql import QueryWriteRejected
 
 from . import Context
 from .base import DatasetteError, View, stream_csv
-from .query_helpers import _ensure_stored_query_execution_permissions, _table_columns
+from .query_helpers import (
+    _block_framing,
+    _ensure_stored_query_execution_permissions,
+    _table_columns,
+)
 from .table_create_alter import _create_table_ui_context
 from .table_extras import (
     QueryExtraContext,
@@ -857,7 +861,8 @@ class QueryView(View):
                 raise DatasetteError("?sql= is required", status=400)
 
             async def fetch_data_for_csv(request, _next=None):
-                results = await db.execute(sql, params, truncate=True)
+                # Reuse the trusted magic parameter values prepared above.
+                results = await db.execute(sql, params_for_query, truncate=True)
                 data = {"rows": results.rows, "columns": results.columns}
                 return data, None, None
 
@@ -1140,6 +1145,8 @@ class QueryView(View):
             assert False, f"Invalid format: {format_}"
         if datasette.cors:
             add_cors_headers(r.headers)
+        if stored_query_write and format_ == "html":
+            _block_framing(r)
         return r
 
 

datasette/views/execute_write.py

@@ -1,6 +1,7 @@
 import re
 from urllib.parse import urlencode
 
+from datasette.database import QueryInterrupted
 from datasette.resources import DatabaseResource
 from datasette.utils import UNSTABLE_API_MESSAGE, sqlite3
 from datasette.utils.asgi import Response
@@ -384,7 +385,7 @@ class ExecuteWriteView(BaseView):
         try:
             execute_write_kwargs = {"request": request}
             cursor = await db.execute_write(sql, params, **execute_write_kwargs)
-        except sqlite3.DatabaseError as ex:
+        except (QueryInterrupted, sqlite3.DatabaseError) as ex:
             message = str(ex)
             if wants_json:
                 return _block_framing(Response.error([message], 400))

datasette/views/row.py

@@ -28,9 +28,11 @@ from datasette.utils import (
     path_with_format,
     path_with_removed_args,
     sqlite3,
+    tilde_decode,
     to_css_class,
 )
 from datasette.utils.asgi import Forbidden, NotFound, PayloadTooLarge, Response
+from datasette.utils.sqlite import check_structured_write_table
 
 from . import Context, from_extra
 from .base import BaseView, DatasetteError, stream_csv
@@ -137,6 +139,12 @@ class RowContext(Context):
     )
 
 
+async def _database_and_table_resource_from_request(datasette, request):
+    db = await datasette.resolve_database(request)
+    table = tilde_decode(request.url_vars["table"])
+    return db, table, TableResource(database=db.name, table=table)
+
+
 class RowView(BaseView):
     name = "row"
 
@@ -263,7 +271,7 @@ class RowView(BaseView):
         if ttl is None or not ttl.isdigit():
             ttl = self.ds.setting("default_cache_ttl")
 
-        return self.set_response_headers(response, ttl)
+        return self.set_response_headers(response, ttl, request)
 
     async def html(self, request, data, extra_template_data, templates):
         extras = {}
@@ -376,36 +384,50 @@ class RowView(BaseView):
             },
         )
 
-    def set_response_headers(self, response, ttl):
+    def set_response_headers(self, response, ttl, request=None):
+        private = getattr(request, "_datasette_private_response", False)
         # Set far-future cache expiry
         if self.ds.cache_headers and response.status == 200:
-            ttl = int(ttl)
-            if ttl == 0:
-                ttl_header = "no-cache"
+            if private:
+                # This response is only visible to the current actor (denied
+                # to anonymous requests), so it must never be stored by a
+                # shared cache/CDN - and ?_ttl= must not override that.
+                response.headers["Cache-Control"] = "private, no-store"
+                response.headers["Vary"] = "Cookie"
             else:
-                ttl_header = f"max-age={ttl}"
-            response.headers["Cache-Control"] = ttl_header
+                ttl = int(ttl)
+                if ttl == 0:
+                    ttl_header = "no-cache"
+                else:
+                    ttl_header = f"max-age={ttl}"
+                response.headers["Cache-Control"] = ttl_header
         response.headers["Referrer-Policy"] = "no-referrer"
         if self.ds.cors:
             add_cors_headers(response.headers)
         return response
 
     async def data(self, request, default_labels=False):
-        resolved = await self.ds.resolve_row(request)
-        db = resolved.db
+        db, table, resource = await _database_and_table_resource_from_request(
+            self.ds, request
+        )
         database = db.name
-        table = resolved.table
-        pk_values = resolved.pk_values
 
-        # Ensure user has permission to view this row
+        # Check the URL resource before resolving the row, so a denied request
+        # cannot distinguish an existing primary key from a missing one.
         visible, private = await self.ds.check_visibility(
             request.actor,
             action="view-table",
-            resource=TableResource(database=database, table=table),
+            resource=resource,
         )
         if not visible:
             raise Forbidden("You do not have permission to view this table")
+        # Record whether this response is private (visible to this actor
+        # only) so set_response_headers() can set appropriate Cache-Control
+        # headers, regardless of which output format ends up being rendered.
+        request._datasette_private_response = private
 
+        resolved = await self.ds.resolve_row(request)
+        pk_values = resolved.pk_values
         results = await resolved.db.execute(
             resolved.sql, resolved.params, truncate=True
         )
@@ -482,8 +504,8 @@ class RowView(BaseView):
             for row in display_rows:
                 for cell in row:
                     if cell["column"] in pk_set:
-                        cell["value"] = markupsafe.Markup(
-                            "<strong>{}</strong>".format(cell["value"])
+                        cell["value"] = markupsafe.Markup("<strong>{}</strong>").format(
+                            cell["value"]
                         )
 
             label_column = await db.label_column_for_table(table) if is_table else None
@@ -556,7 +578,7 @@ class RowView(BaseView):
                 "private": private,
                 "columns": reordered_columns,
                 "foreign_key_tables": await self.foreign_key_tables(
-                    database, table, pk_values
+                    database, table, pk_values, actor=request.actor
                 ),
                 "database_color": db.color,
                 "display_columns": display_columns,
@@ -633,12 +655,23 @@ class RowView(BaseView):
             ),
         )
 
-    async def foreign_key_tables(self, database, table, pk_values):
+    async def foreign_key_tables(self, database, table, pk_values, *, actor):
         if len(pk_values) != 1:
             return []
         db = self.ds.databases[database]
         all_foreign_keys = await db.get_all_foreign_keys()
-        foreign_keys = all_foreign_keys[table]["incoming"]
+        foreign_keys = []
+        table_permissions = {}
+        for fk in all_foreign_keys[table]["incoming"]:
+            other_table = fk["other_table"]
+            if other_table not in table_permissions:
+                table_permissions[other_table] = await self.ds.allowed(
+                    action="view-table",
+                    resource=TableResource(database=database, table=other_table),
+                    actor=actor,
+                )
+            if table_permissions[other_table]:
+                foreign_keys.append(fk)
         if len(foreign_keys) == 0:
             return []
 
@@ -695,9 +728,24 @@ def _truncated_row_flash_label(label):
     return label[: ROW_FLASH_LABEL_MAX_LENGTH - 1] + "\u2026"
 
 
-async def _row_flash_message(db, action, resolved, row=None):
+async def _row_flash_message(
+    datasette, request, action, resolved, row=None, *, refresh_row=False
+):
     pk_label = ", ".join(resolved.pk_values)
-    label_column = await db.label_column_for_table(resolved.table)
+    # Mutation permission does not grant access to stored row labels.
+    if not await datasette.allowed(
+        action="view-table",
+        resource=TableResource(database=resolved.db.name, table=resolved.table),
+        actor=request.actor,
+    ):
+        return f"{action} row {pk_label}"
+
+    if refresh_row and row is None:
+        results = await resolved.db.execute(
+            resolved.sql, resolved.params, truncate=True
+        )
+        row = results.first()
+    label_column = await resolved.db.label_column_for_table(resolved.table)
     label = row_label_from_label_column(row or resolved.row, label_column)
     if label:
         label = _truncated_row_flash_label(label)
@@ -710,22 +758,28 @@ async def _resolve_row_and_check_permission(datasette, request, permission):
     from datasette.app import DatabaseNotFound, RowNotFound, TableNotFound
 
     try:
-        resolved = await datasette.resolve_row(request)
+        _, _, resource = await _database_and_table_resource_from_request(
+            datasette, request
+        )
     except DatabaseNotFound as e:
         return False, Response.error([f"Database not found: {e.database_name}"], 404)
+
+    # Check the URL resource before resolving the row, so a denied request
+    # cannot distinguish an existing primary key from a missing one.
+    if not await datasette.allowed(
+        action=permission,
+        resource=resource,
+        actor=request.actor,
+    ):
+        return False, Response.error(["Permission denied"], 403)
+
+    try:
+        resolved = await datasette.resolve_row(request)
     except TableNotFound as e:
         return False, Response.error([f"Table not found: {e.table}"], 404)
     except RowNotFound as e:
         return False, Response.error([f"Record not found: {e.pk_values}"], 404)
 
-    # Ensure user has permission to delete this row
-    if not await datasette.allowed(
-        action=permission,
-        resource=TableResource(database=resolved.db.name, table=resolved.table),
-        actor=request.actor,
-    ):
-        return False, Response.error(["Permission denied"], 403)
-
     return True, resolved
 
 
@@ -744,6 +798,7 @@ class RowDeleteView(BaseView):
 
         # Delete table
         def delete_row(conn):
+            check_structured_write_table(conn, resolved.table)
             sqlite_utils.Database(conn)[resolved.table].delete(resolved.pk_values)
 
         try:
@@ -765,7 +820,7 @@ class RowDeleteView(BaseView):
             table_url = self.ds.urls.table(resolved.db.name, resolved.table)
             self.ds.add_message(
                 request,
-                await _row_flash_message(resolved.db, "Deleted", resolved),
+                await _row_flash_message(self.ds, request, "Deleted", resolved),
                 self.ds.INFO,
             )
             return Response.json({"ok": True, "redirect": str(table_url)}, status=200)
@@ -826,6 +881,7 @@ class RowUpdateView(BaseView):
             return Response.error(["Permission denied for alter-table"], 403)
 
         def update_row(conn):
+            check_structured_write_table(conn, resolved.table)
             sqlite_utils.Database(conn)[resolved.table].update(
                 resolved.pk_values, update, alter=alter
             )
@@ -838,7 +894,14 @@ class RowUpdateView(BaseView):
 
         result = {"ok": True}
         returned_row = None
-        if data.get("return"):
+        # Only read back and disclose the stored row if the actor is also
+        # allowed to view this table - update-row alone must not be usable
+        # to read data the actor cannot otherwise see.
+        if data.get("return") and await self.ds.allowed(
+            action="view-table",
+            resource=TableResource(database=resolved.db.name, table=resolved.table),
+            actor=request.actor,
+        ):
             results = await resolved.db.execute(
                 resolved.sql, resolved.params, truncate=True
             )
@@ -855,16 +918,15 @@ class RowUpdateView(BaseView):
         )
 
         if request.args.get("_message"):
-            message_row = returned_row
-            if message_row is None:
-                results = await resolved.db.execute(
-                    resolved.sql, resolved.params, truncate=True
-                )
-                message_row = results.first()
             self.ds.add_message(
                 request,
                 await _row_flash_message(
-                    resolved.db, "Updated", resolved, row=message_row
+                    self.ds,
+                    request,
+                    "Updated",
+                    resolved,
+                    row=returned_row,
+                    refresh_row=True,
                 ),
                 self.ds.INFO,
             )

datasette/views/special.py

@@ -311,6 +311,7 @@ class AllowedResourcesView(BaseView):
     has_json_alternate = False
 
     async def get(self, request):
+        await self.ds.ensure_permission(action="view-instance", actor=request.actor)
         await self.ds.refresh_schemas()
 
         # Check if user has permissions-debug (to show sensitive fields)
@@ -796,6 +797,8 @@ class CreateTokenView(BaseView):
             raise Forbidden(
                 "Token authentication cannot be used to create additional tokens"
             )
+        if "_r" in request.actor:
+            raise Forbidden("Restricted actors cannot create API tokens")
 
     async def shared(self, request):
         self.check_permission(request)
@@ -873,6 +876,11 @@ class CreateTokenView(BaseView):
                 else:
                     errors.append("Invalid expire duration unit")
 
+        if errors:
+            context = await self.shared(request)
+            context["errors"] = errors
+            return await self.render(["create_token.html"], request, context)
+
         # Are there any restrictions?
         from datasette.tokens import TokenRestrictions
 
@@ -1261,14 +1269,21 @@ class SchemaBaseView(BaseView):
 
     has_json_alternate = False
 
-    async def get_database_schema(self, database_name):
+    async def get_database_schema(self, database_name, actor):
         """Get schema SQL for a database."""
         db = self.ds.databases[database_name]
-        result = await db.execute(
-            "select group_concat(sql, ';' || CHAR(10)) as schema from sqlite_master where sql is not null"
+        allowed_tables_page = await self.ds.allowed_resources(
+            "view-table", actor, parent=database_name
+        )
+        allowed_table_names = {
+            resource.child async for resource in allowed_tables_page.all()
+        }
+        result = await db.execute(
+            "select tbl_name, sql from sqlite_master where sql is not null"
+        )
+        return ";\n".join(
+            row["sql"] for row in result.rows if row["tbl_name"] in allowed_table_names
         )
-        row = result.first()
-        return row["schema"] if row and row["schema"] else ""
 
     def format_json_response(self, data):
         """Format data as JSON response with CORS headers if needed."""
@@ -1330,7 +1345,7 @@ class InstanceSchemaView(SchemaBaseView):
         # Get schema for each database
         schemas = []
         for database_name in allowed_databases:
-            schema = await self.get_database_schema(database_name)
+            schema = await self.get_database_schema(database_name, request.actor)
             schemas.append({"database": database_name, "schema": schema})
 
         if format_ == "json":
@@ -1371,7 +1386,7 @@ class DatabaseSchemaView(SchemaBaseView):
         if database_name not in self.ds.databases:
             return self.format_error_response("Database not found", format_)
 
-        schema = await self.get_database_schema(database_name)
+        schema = await self.get_database_schema(database_name, request.actor)
 
         if format_ == "json":
             return self.format_json_response(
@@ -1410,7 +1425,8 @@ class TableSchemaView(SchemaBaseView):
         # Get schema for the table
         db = self.ds.databases[database_name]
         result = await db.execute(
-            "select sql from sqlite_master where name = ? and sql is not null",
+            "select sql from sqlite_master where name = ? "
+            "and type in ('table', 'view') and sql is not null",
             [table_name],
         )
         row = result.first()

datasette/views/stored_queries.py

@@ -279,7 +279,7 @@ class QueryCreateView(BaseView):
             ),
         )
         response.status = status
-        return response
+        return _block_framing(response)
 
     async def get(self, request):
         db = await self.ds.resolve_database(request)
@@ -527,7 +527,7 @@ class QueryEditView(BaseView):
             ),
         )
         response.status = status
-        return response
+        return _block_framing(response)
 
     async def get(self, request):
         db, query_name, existing = await self._load(request)
@@ -639,15 +639,17 @@ class QueryDeleteView(BaseView):
             return Response.error(
                 ["Trusted queries cannot be deleted using the API"], 403
             )
-        return await self.render(
-            ["query_delete.html"],
-            request,
-            {
-                "database": db.name,
-                "database_color": db.color,
-                "query": stored_query_to_dict(existing),
-                "query_url": self.ds.urls.table(db.name, query_name),
-            },
+        return _block_framing(
+            await self.render(
+                ["query_delete.html"],
+                request,
+                {
+                    "database": db.name,
+                    "database_color": db.color,
+                    "query": stored_query_to_dict(existing),
+                    "query_url": self.ds.urls.table(db.name, query_name),
+                },
+            )
         )
 
     async def post(self, request):

datasette/views/table.py

@@ -57,6 +57,7 @@ from datasette.utils.asgi import (
     Request,
     Response,
 )
+from datasette.utils.sqlite import check_structured_write_table
 
 from . import Context, from_extra
 from .base import BaseView, DatasetteError, stream_csv
@@ -1126,6 +1127,7 @@ class TableInsertView(BaseView):
             row_pk_values_for_later = [tuple(row[pk] for pk in pks) for row in rows]
 
         def insert_or_upsert_rows(conn):
+            check_structured_write_table(conn, table_name)
             table = sqlite_utils.Database(conn)[table_name]
             kwargs = {}
             if upsert:
@@ -1157,17 +1159,32 @@ class TableInsertView(BaseView):
             # TODO: narrow to expected write errors so Datasette bugs surface as 500s
             return Response.error([str(e)])
         result = {"ok": True}
+        # Only read back and disclose stored rows if the actor is also
+        # allowed to view this table - insert-row/update-row alone must
+        # not be usable to read data the actor cannot otherwise see.
+        if should_return and not await self.ds.allowed(
+            action="view-table",
+            resource=TableResource(database=database_name, table=table_name),
+            actor=request.actor,
+        ):
+            should_return = False
         if should_return:
             if upsert:
                 # Fetch based on initial input IDs
                 where_clause = " OR ".join(
-                    ["({})".format(" AND ".join(f"{pk} = ?" for pk in pks))]
+                    [
+                        "({})".format(
+                            " AND ".join(f"{escape_sqlite(pk)} = ?" for pk in pks)
+                        )
+                    ]
                     * len(row_pk_values_for_later)
                 )
                 args = list(itertools.chain.from_iterable(row_pk_values_for_later))
                 fetched_rows = await db.execute(
-                    "select {}* from [{}] where {}".format(
-                        "rowid, " if pks == ["rowid"] else "", table_name, where_clause
+                    "select {}* from {} where {}".format(
+                        "rowid, " if pks == ["rowid"] else "",
+                        escape_sqlite(table_name),
+                        where_clause,
                     ),
                     args,
                 )
@@ -1382,7 +1399,9 @@ class TableDropView(BaseView):
                     "database": database_name,
                     "table": table_name,
                     "row_count": (
-                        await db.execute(f"select count(*) from [{table_name}]")
+                        await db.execute(
+                            f"select count(*) from {escape_sqlite(table_name)}"
+                        )
                     ).single_value(),
                     "message": 'Pass "confirm": true to confirm',
                 },
@@ -1695,13 +1714,22 @@ async def table_view(datasette, request):
     if ttl is None or not ttl.isdigit():
         ttl = datasette.setting("default_cache_ttl")
 
+    private = getattr(request, "_datasette_private_response", False)
+
     if datasette.cache_headers and response.status == 200:
-        ttl = int(ttl)
-        if ttl == 0:
-            ttl_header = "no-cache"
+        if private:
+            # This response is only visible to the current actor (denied to
+            # anonymous requests), so it must never be stored by a shared
+            # cache/CDN - and ?_ttl= must not be able to override that.
+            response.headers["Cache-Control"] = "private, no-store"
+            response.headers["Vary"] = "Cookie"
         else:
-            ttl_header = f"max-age={ttl}"
-        response.headers["Cache-Control"] = ttl_header
+            ttl = int(ttl)
+            if ttl == 0:
+                ttl_header = "no-cache"
+            else:
+                ttl_header = f"max-age={ttl}"
+            response.headers["Cache-Control"] = ttl_header
 
     # Referrer policy
     response.headers["Referrer-Policy"] = "no-referrer"
@@ -1949,6 +1977,10 @@ async def table_view_data(
     )
     if not visible:
         raise Forbidden("You do not have permission to view this table")
+    # Record whether this response is private (visible to this actor only)
+    # so the outer table_view() can set appropriate Cache-Control headers,
+    # regardless of which output format ends up being rendered.
+    request._datasette_private_response = private
 
     # Redirect based on request.args, if necessary
     redirect_response = await _redirect_if_needed(datasette, request, resolved)
@@ -2432,9 +2464,12 @@ async def _next_value_and_url(
             except IndexError:
                 # sort/sort_desc column missing from SELECT - look up value by PK instead
                 prefix_where_clause = " and ".join(
-                    f"[{pk}] = :pk{i}" for i, pk in enumerate(pks)
+                    f"{escape_sqlite(pk)} = :pk{i}" for i, pk in enumerate(pks)
+                )
+                prefix_lookup_sql = (
+                    f"select {escape_sqlite(sort or sort_desc)} "
+                    f"from {escape_sqlite(table_name)} where {prefix_where_clause}"
                 )
-                prefix_lookup_sql = f"select [{sort or sort_desc}] from [{table_name}] where {prefix_where_clause}"
                 prefix = (
                     await db.execute(
                         prefix_lookup_sql,

datasette/views/table_create_alter.py

@@ -27,7 +27,15 @@ from datasette.utils import (
     table_column_details,
 )
 from datasette.utils.asgi import NotFound, PayloadTooLarge, Response
-from datasette.utils.sqlite import sqlite_hidden_table_names
+from datasette.utils.permissions import (
+    SKIP_PERMISSION_CHECKS,
+    gather_permission_sql_from_hooks,
+    resolve_permissions_with_candidates,
+)
+from datasette.utils.sqlite import (
+    check_structured_write_table,
+    sqlite_hidden_table_names,
+)
 
 from .base import BaseView
 
@@ -122,6 +130,30 @@ def _public_foreign_key_target(target):
     }
 
 
+async def _filter_visible_foreign_key_targets(datasette, actor, database_name, targets):
+    if not targets:
+        return []
+
+    permission_sqls = await gather_permission_sql_from_hooks(
+        datasette=datasette,
+        actor=actor,
+        action="view-table",
+    )
+    if permission_sqls is SKIP_PERMISSION_CHECKS:
+        return targets
+
+    candidate_tables = list(dict.fromkeys(target["fk_table"] for target in targets))
+    permission_rows = await resolve_permissions_with_candidates(
+        datasette.get_internal_database(),
+        actor,
+        permission_sqls,
+        [(database_name, table_name) for table_name in candidate_tables],
+        "view-table",
+    )
+    visible_tables = {row["child"] for row in permission_rows if bool(row["allow"])}
+    return [target for target in targets if target["fk_table"] in visible_tables]
+
+
 def _singular(name):
     if name.endswith("ies") and len(name) > 3:
         return name[:-3] + "y"
@@ -821,16 +853,18 @@ class TableCreateView(BaseView):
         ignore = create_request.ignore
         replace = create_request.replace
 
+        table_name = create_request.table
+        table_exists = await db.table_exists(table_name)
+        table_resource = TableResource(database=database_name, table=table_name)
+
         # Replacing rows requires update-row permission
         if replace and not await self.ds.allowed(
             action="update-row",
-            resource=DatabaseResource(database=database_name),
+            resource=table_resource,
             actor=request.actor,
         ):
             return Response.error(["Permission denied: need update-row"], 403)
 
-        table_name = create_request.table
-        table_exists = await db.table_exists(table_name)
         columns = create_request.columns
         rows = create_request.rows_list
 
@@ -838,7 +872,7 @@ class TableCreateView(BaseView):
             # Must have insert-row permission
             if not await self.ds.allowed(
                 action="insert-row",
-                resource=DatabaseResource(database=database_name),
+                resource=table_resource,
                 actor=request.actor,
             ):
                 return Response.error(["Permission denied: need insert-row"], 403)
@@ -857,7 +891,7 @@ class TableCreateView(BaseView):
                 if create_request.alter:
                     if not await self.ds.allowed(
                         action="alter-table",
-                        resource=DatabaseResource(database=database_name),
+                        resource=table_resource,
                         actor=request.actor,
                     ):
                         return Response.error(
@@ -893,6 +927,7 @@ class TableCreateView(BaseView):
             )
 
         def create_table(conn):
+            check_structured_write_table(conn, table_name, allow_missing=True)
             db_for_write = sqlite_utils.Database(conn)
             table = db_for_write[table_name]
             if rows:
@@ -1012,6 +1047,9 @@ class DatabaseForeignKeyTargetsView(BaseView):
             for target in (await db.execute(FOREIGN_KEY_TARGETS_SQL)).dicts()
             if target["fk_table"] not in hidden_tables
         ]
+        targets = await _filter_visible_foreign_key_targets(
+            self.ds, request.actor, database_name, targets
+        )
         return Response.json(
             {
                 "ok": True,
@@ -1050,6 +1088,15 @@ class TableForeignKeySuggestionsView(BaseView):
         source_columns, targets, current_by_column = await db.execute_fn(
             lambda conn: _foreign_key_suggestion_metadata(conn, table_name)
         )
+        targets = await _filter_visible_foreign_key_targets(
+            self.ds, request.actor, database_name, targets
+        )
+        visible_target_tables = {target["fk_table"] for target in targets}
+        current_by_column = {
+            column: current
+            for column, current in current_by_column.items()
+            if current["fk_table"] in visible_target_tables
+        }
 
         columns = []
         options_by_column = {}

datasette/views/table_extras.py

@@ -1206,7 +1206,10 @@ class ForeignKeyTablesExtra(Extra):
 
     async def resolve(self, context):
         return await context.foreign_key_tables(
-            context.database_name, context.table_name, context.pk_values
+            context.database_name,
+            context.table_name,
+            context.pk_values,
+            actor=context.request.actor,
         )
 
 

datasette/write_sql.py

@@ -83,6 +83,22 @@ def decision_for_write_sql_operation(
             )
     if operation.operation == "function":
         return IgnoreWriteSqlOperation("SQL function")
+    if (
+        operation.operation == "read"
+        and operation.target_type == "table"
+        and operation.table is not None
+        and operation.table_kind is None
+        and operation.table.lower().startswith("pragma_")
+    ):
+        # Eponymous table-valued PRAGMA functions (e.g. pragma_table_info("secret"))
+        # report a read of the synthetic "pragma_table_info" table, not of the
+        # table passed as an argument. That means a view-table denial on the real
+        # table is never consulted, so these could otherwise be used to read
+        # schema metadata (column names, table lists, ...) for tables the actor
+        # is not allowed to view. Reject them outright in untrusted write SQL,
+        # including inside CREATE VIEW bodies (whose reads are discovered here
+        # via the rolled-back dependency-read analysis above).
+        return UnsupportedWriteSqlOperation(unsupported_message)
     if (
         operation.operation == "read"
         and operation.target_type == "table"

docs/authentication.rst

@@ -158,6 +158,15 @@ Datasette resolves matching rules from most specific to least specific:
 
 This means a resource-level allow can provide an exception to a parent-level deny. It also means that two plugins which disagree at the same level resolve to deny.
 
+For table and view permissions, resource names use SQLite's case-insensitive
+identifier matching: ``Secret``, ``secret`` and ``SECRET`` identify the same
+table. This applies to configuration rules, plugin rules and token restrictions.
+Only ASCII letters are case-insensitive; non-ASCII characters remain distinct.
+Conflicting rules for different spellings of the same name follow the usual
+deny-wins rule at the same scope. Names retain their original spelling in
+resource listings and permission explanations. Database names, stored query
+names and other resource types remain case-sensitive.
+
 .. list-table:: Permission rule examples
    :header-rows: 1
 
@@ -182,6 +191,18 @@ This means a resource-level allow can provide an exception to a parent-level den
 
 The built-in public defaults are global allow rules for actions such as ``view-instance``, ``view-database`` and ``view-table``. They follow the same precedence rules as configuration and plugin rules. The ``--default-deny`` option prevents Datasette from contributing those default allow rules.
 
+The built-in ``datasette.default_permissions.sqlite_statistics`` plugin denies
+``view-table`` for ``sqlite_stat1``, ``sqlite_stat2``, ``sqlite_stat3`` and
+``sqlite_stat4``. These table-level denials also apply to root users and take
+precedence over configuration or plugin allow rules at the same scope.
+This controls table access and listings, without changing ``execute-sql`` or
+SQLite's internal use of statistics.
+
+A plugin can replace this policy by unregistering
+``datasette.default_permissions.sqlite_statistics`` through ``datasette.pm``
+and registering its own permission hook. Plugin registration is process-wide:
+replacing this policy affects every Datasette instance in that process.
+
 Datasette performs checks using :ref:`datasette_allowed`, which accepts keyword arguments for ``action``, ``resource`` and an optional ``actor``.
 
 ``resource`` should be an instance of the appropriate ``Resource`` subclass from :mod:`datasette.resources`—for example ``InstanceResource()``, ``DatabaseResource(database="...``)`` or ``TableResource(database="...", table="...")``. This defaults to ``InstanceResource()`` if not specified.
@@ -771,6 +792,8 @@ Datasette defaults to allowing any site visitor to execute their own custom SQL
 
 Access to this ability is controlled by the :ref:`actions_execute_sql` permission.
 
+This permission does not apply to structured table-browsing operations where Datasette constructs the SQL, such as sorting, column filters and :ref:`facets`. Faceting is controlled separately by the :ref:`setting_allow_facet` setting.
+
 The easiest way to disable arbitrary SQL queries is using the :ref:`default_allow_sql setting <setting_default_allow_sql>` when you first start Datasette running.
 
 You can alternatively use an ``"allow_sql"`` block to control who is allowed to execute arbitrary SQL queries.
@@ -1359,6 +1382,12 @@ view-table
 
 Actor is allowed to view a table (or view) page, e.g. https://latest.datasette.io/fixtures/complex_foreign_keys
 
+Derived implementation tables require access to their immediate source: FTS and RTree shadow tables require access to their virtual table, external-content FTS tables require access to their content table, and FTS vocabulary tables (``fts5vocab`` and ``fts4aux``) require access to their FTS table. The derived table's own permission rules also apply.
+
+Access is always denied if the source table is itself derived, or if a vocabulary table's source cannot be identified.
+
+The same rules apply to individual permission checks and table listings, including whether they are private. If a database error prevents dependency discovery, the check or listing fails with an error instead of ignoring the dependencies. Failed discovery results are not cached, so later checks can retry.
+
 ``resource`` - ``datasette.resources.TableResource(database, table)``
     ``database`` is the name of the database (string)
 
@@ -1521,6 +1550,8 @@ execute-sql
 
 Actor is allowed to run arbitrary read-only SQL queries against a specific database using the :ref:`custom SQL query page <pages_custom_sql_queries>`, e.g. https://latest.datasette.io/fixtures/-/query?sql=select+100
 
+This action also controls raw SQL supplied using ``?_where=``. It does not control structured table-browsing features such as :ref:`facets`, which use SQL generated by Datasette and are controlled by :ref:`setting_allow_facet`.
+
 ``resource`` - ``datasette.resources.DatabaseResource(database)``
     ``database`` is the name of the database (string)
 

docs/changelog.rst

@@ -4,6 +4,48 @@
 Changelog
 =========
 
+.. _v1_0_a39:
+
+1.0a39 (2026-09-10)
+-------------------
+
+This alpha release includes security fixes for permissions, SQL construction, HTML rendering, authentication and caching, plus improvements to application startup and write execution.
+
+See `0.65.4 <https://docs.datasette.io/en/stable/changelog.html#v0-65-4>`__ for fixes that have been backported to the stable 0.65.x branch.
+
+The Datasette blog `has more details on these releases <https://datasette.io/blog/2026/september-security-releases/>`__.
+
+Some of the security fixes include:
+
+- Table and view permission checks now take SQLite's case-insensitive names into account. See :ref:`authentication_permissions_explained`.
+- Viewing a full-text search index table now checks you have permission to view the table from which it draws its content.
+- Viewing SQLite statistics tables (``sqlite_stat1`` through ``sqlite_stat4``) is now denied by a default.
+- Table schema display now obeys the ``view-table`` permission.
+- Table filters using ``?_through=`` require permission to view the intermediate table.
+- Foreign-key target and suggestion APIs, incoming foreign-key relationships and their row counts now respect ``view-table`` permission.
+- Row endpoints check permissions before resolving primary keys, to avoid revealing the existence of an otherwise invisible primary key.
+- Improved permission checks for the create-table API. See :ref:`json_api_write`.
+- The write SQL interface now checks ``view-table`` permission for tables referenced by ``CREATE VIEW`` statements.
+- Fixed SQL identifier escaping for column names from untrusted database schemas.
+- Fixed HTML escaping for column names from untrusted database schemas.
+- URL columns now render links only for validated HTTP or HTTPS URLs.
+- Private and personalized dynamic responses now use ``Cache-Control: private, no-store``. Anonymous dynamic responses vary by ``Cookie`` and ``Authorization``.
+- Actor cookies now respect ``expire_after``.
+- Restricted actors can no longer create API tokens.
+- Stored-query create, edit and delete forms now block framing to prevent clickjacking.
+- Configuration secret redaction now matches key names case-insensitively.
+- SQLite extension loading is disabled after extensions supplied using ``--load-extension`` have been loaded.
+
+Other improvements and fixes
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+- :ref:`db.execute_write() <database_execute_write>` now has a default execution time limit of 2,000ms. Plugins can override this using ``time_limit_ms=`` or disable it using ``time_limit_ms=None``. This limit is independent of the ``sql_time_limit_ms`` setting for read queries.
+- Application startup now runs through ASGI lifespan events before requests are accepted, with a first-request fallback for hosts without lifespan support. Thanks, `Alex Garcia <https://github.com/asg017>`__. (:pr:`2887`)
+- ``datasette serve`` now runs startup hooks and Uvicorn on the same event loop, preserving background tasks started by plugins. The minimum Uvicorn version is now 0.29. Thanks, `Alex Garcia <https://github.com/asg017>`__. (:pr:`2886`)
+- Non-blocking writes using ``execute_write_fn(..., block=False)`` now return a distinct task UUID for every call and work correctly with ``num_sql_threads=0``. Thanks, `Zain Dana Harper <https://github.com/HarperZ9>`__. (:issue:`2860`, :issue:`2859`)
+- Dropping a table now disables its full-text search index first. (:issue:`2874`)
+- Fixed ``CREATE VIEW`` SQL analysis on Python 3.10.
+
 .. _v1_0_a38:
 
 1.0a38 (2026-08-06)

docs/facets.rst

@@ -14,6 +14,8 @@ Here's `an example <https://congress-legislators.datasettes.com/legislators/legi
 
 Facets can be specified in two ways: using query string parameters, or in ``metadata.json`` configuration for the table.
 
+Facet queries are generated by Datasette and summarize rows the actor already has permission to view. They do not require the :ref:`actions_execute_sql` permission. Use the :ref:`setting_allow_facet` setting to control whether users can request facets using query string parameters.
+
 Facets in query strings
 -----------------------
 

docs/full_text_search.rst

@@ -52,11 +52,11 @@ Configuring full-text search for a table or view
 
 If a table has a corresponding FTS table set up using the ``content=`` argument to ``CREATE VIRTUAL TABLE`` shown below, Datasette will detect it automatically and add a search interface to the table page for that table.
 
-You can also manually configure which table should be used for full-text search using query string parameters or table configuration in ``datasette.yaml`` (see :ref:`table_configuration_fts`). You can set the associated FTS table for a specific table and you can also set one for a view - if you do that, the page for that SQL view will offer a search option.
+You can also manually configure which table should be used for full-text search using table configuration in ``datasette.yaml`` (see :ref:`table_configuration_fts`). You can set the associated FTS table for a specific table and you can also set one for a view - if you do that, the page for that SQL view will offer a search option.
 
-Use ``?_fts_table=x`` to over-ride the FTS table for a specific page. If the primary key was something other than ``rowid`` you can use ``?_fts_pk=col`` to set that as well. This is particularly useful for views, for example:
+The legacy ``?_fts_table=x`` and ``?_fts_pk=col`` query string parameters are accepted only if they exactly match the configured or automatically detected FTS mapping. They cannot be used to select a different FTS table or primary key. This prevents a public table from being used to probe the contents of a private FTS table.
 
-https://latest.datasette.io/fixtures/searchable_view?_fts_table=searchable_fts&_fts_pk=pk
+Searching also requires the current actor to have ``view-table`` permission for the FTS table itself, in addition to permission to view the table or view being searched.
 
 The ``fts_table`` metadata property can be used to specify an associated FTS table. If the primary key column in your table which was used to populate the FTS table is something other than ``rowid``, you can specify the column to use with the ``fts_pk`` property.
 

docs/internals.rst

@@ -2023,8 +2023,8 @@ Example usage:
 
 .. _database_execute_write:
 
-await db.execute_write(sql, params=None, block=True, request=None, return_all=False, returning_limit=10, transaction=True)
---------------------------------------------------------------------------------------------------------------------------
+await db.execute_write(sql, params=None, block=True, request=None, return_all=False, returning_limit=10, transaction=True, time_limit_ms=2000)
+----------------------------------------------------------------------------------------------------------------------------------------------
 
 SQLite only allows one database connection to write at a time. Datasette handles this for you by maintaining a queue of writes to be executed against a given database. Plugins can submit write operations to this queue and they will be executed in the order in which they are received.
 
@@ -2063,6 +2063,13 @@ Each call to ``execute_write()`` will be executed inside a transaction. Pass
 ``transaction=False`` for statements such as ``VACUUM`` that cannot run inside
 a transaction.
 
+Write statements have a default time limit of 2,000ms. Pass a different value
+using ``time_limit_ms=`` or use ``time_limit_ms=None`` to allow the statement to
+run without a time limit.
+
+This write limit is independent of the ``sql_time_limit_ms`` setting used for
+read queries. Changing that setting does not change the default write limit.
+
 .. _database_execute_write_script:
 
 await db.execute_write_script(sql, block=True)

docs/json_api.rst

@@ -1661,6 +1661,8 @@ The request body is always parsed as JSON, regardless of the request's ``Content
 
 The row-based write APIs can write :ref:`binary values in JSON <binary_json_format>` using Datasette's Base64 representation for BLOB data.
 
+Structured inserts, upserts, updates and deletes only support ordinary SQLite tables. Virtual tables and their internal shadow tables are rejected, including when adding rows to an existing table through the create-table API. Writes to ordinary content tables can still update full-text search indexes through configured triggers.
+
 .. _ExecuteWriteView:
 
 Executing write SQL

docs/plugins.rst

@@ -261,6 +261,15 @@ If you run ``datasette plugins --all`` it will include default plugins that ship
                 "permission_resources_sql"
             ]
         },
+        {
+            "name": "datasette.default_permissions.sqlite_statistics",
+            "static": false,
+            "templates": false,
+            "version": null,
+            "hooks": [
+                "permission_resources_sql"
+            ]
+        },
         {
             "name": "datasette.default_permissions.tokens",
             "static": false,

docs/settings.rst

@@ -71,6 +71,8 @@ Should users be able to execute arbitrary SQL queries by default?
 
 Setting this to ``off`` causes permission checks for :ref:`actions_execute_sql` to fail by default.
 
+This setting controls the ability to submit arbitrary SQL. It does not disable structured table-browsing features that use SQL generated by Datasette, such as sorting, column filters and :ref:`facets`. Use :ref:`setting_allow_facet` to control whether users can request facets.
+
 ::
 
     datasette mydatabase.db --setting default_allow_sql off
@@ -254,6 +256,8 @@ Default HTTP caching max-age header in seconds, used for ``Cache-Control: max-ag
 
     datasette mydatabase.db --setting default_cache_ttl 60
 
+Dynamic responses for authenticated actors, requests with cookies or an ``Authorization`` header, and responses that set cookies use ``Cache-Control: private, no-store``. This takes precedence over ``default_cache_ttl`` and ``?_ttl=``, even when cache headers are otherwise disabled. Anonymous dynamic responses vary by ``Cookie`` and ``Authorization``. Static assets retain their own cache policy.
+
 .. _setting_cache_size_kb:
 
 cache_size_kb

tests/test_api.py

@@ -5,6 +5,7 @@ import pytest
 
 from datasette.app import Datasette
 from datasette.plugins import DEFAULT_PLUGINS
+from datasette.resources import DatabaseResource, TableResource
 from datasette.utils import UNSTABLE_API_MESSAGE, escape_sqlite, tilde_encode
 from datasette.utils.sqlite import sqlite_version
 from datasette.version import __version__
@@ -101,14 +102,11 @@ async def test_database_page(ds_client):
         "tags",
     }
 
-    # Expected hidden tables
+    # The external-content index is visible, but its shadow tables need a
+    # second dependency hop and are excluded by the one-hop permission policy.
     expected_hidden_tables = {
         "no_primary_key",
         "searchable_fts",
-        "searchable_fts_config",
-        "searchable_fts_data",
-        "searchable_fts_docsize",
-        "searchable_fts_idx",
     }
 
     # Verify all expected tables exist
@@ -458,6 +456,67 @@ async def test_row_foreign_key_tables(ds_client):
     ]
 
 
+@pytest.mark.asyncio
+async def test_row_foreign_key_tables_omit_denied_tables(request):
+    actor = {"id": "reader"}
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "databases": {
+                "data": {
+                    "tables": {
+                        "parents": {"permissions": {"view-table": True}},
+                        "private_children": {"permissions": {"view-table": False}},
+                    }
+                }
+            }
+        },
+    )
+    request.addfinalizer(ds.close)
+    db = ds.add_memory_database("fk_count_leak", name="data")
+    await db.execute_write("create table parents (id integer primary key, name text)")
+    await db.execute_write("""
+        create table private_children (
+            id integer primary key,
+            parent_id integer references parents(id)
+        )
+    """)
+    await db.execute_write("insert into parents values (1, 'Public parent')")
+    await db.execute_write("""
+        insert into private_children (id, parent_id) values
+            (1, 1),
+            (2, 1),
+            (3, 1)
+    """)
+    await ds.invoke_startup()
+
+    parent = TableResource(database="data", table="parents")
+    private_children = TableResource(database="data", table="private_children")
+    assert await ds.allowed(action="view-table", resource=parent, actor=actor)
+    assert not await ds.allowed(
+        action="view-table", resource=private_children, actor=actor
+    )
+    assert not await ds.allowed(
+        action="execute-sql",
+        resource=DatabaseResource(database="data"),
+        actor=actor,
+    )
+
+    direct_child = await ds.client.get("/data/private_children.json", actor=actor)
+    assert direct_child.status_code == 403
+    parent_response = await ds.client.get(
+        "/data/parents/1.json?_extra=foreign_key_tables", actor=actor
+    )
+    assert parent_response.status_code == 200
+
+    foreign_key_tables = parent_response.json().get("foreign_key_tables", [])
+    assert foreign_key_tables == [], (
+        "denied child table name, foreign-key column, and row count disclosed: "
+        f"{foreign_key_tables}"
+    )
+
+
 @pytest.mark.asyncio
 async def test_row_extras(ds_client):
     response = await ds_client.get(
@@ -894,10 +953,7 @@ async def test_hidden_sqlite_stat1_table():
     await db.execute_write("analyze")
     data = (await ds.client.get("/db.json?_show_hidden=1")).json()
     tables = [(t["name"], t["hidden"]) for t in data["tables"]]
-    assert tables in (
-        [("normal", False), ("sqlite_stat1", True)],
-        [("normal", False), ("sqlite_stat1", True), ("sqlite_stat4", True)],
-    )
+    assert tables == [("normal", False)]
 
 
 @pytest.mark.asyncio

tests/test_api_write.py

@@ -68,6 +68,82 @@ BASE64_WRITE_API_VALUE = {"$base64": True, "encoded": "AAEC/f7/"}
 BASE64_WRITE_API_LITERAL = '{"$base64": true, "encoded": "AAEC/f7/"}'
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_fallback", (False, True))
+@pytest.mark.parametrize(
+    "operation", ("insert", "upsert", "update", "delete", "create", "create_uppercase")
+)
+@pytest.mark.parametrize(
+    "module,definition,values,shadow_suffix",
+    (
+        ("fts5", "body", "'original'", "_content"),
+        ("fts4", "body", "'original'", "_content"),
+        ("rtree", "id, minx, maxx", "1, 0, 1", "_rowid"),
+    ),
+)
+@pytest.mark.parametrize("shadow", (False, True))
+async def test_structured_writes_require_ordinary_tables(
+    ds_write,
+    monkeypatch,
+    use_fallback,
+    operation,
+    module,
+    definition,
+    values,
+    shadow_suffix,
+    shadow,
+):
+    if use_fallback:
+        monkeypatch.setattr("datasette.utils.sqlite.supports_table_list", lambda: False)
+    db = ds_write.get_database("data")
+    await db.execute_write(f"create virtual table indexed using {module}({definition})")
+    await db.execute_write(f"insert into indexed values ({values})")
+    table = "indexed" + (shadow_suffix if shadow else "")
+    row = (await db.execute(f"select rowid, * from {escape_sqlite(table)}")).dicts()[0]
+    pks = await db.primary_keys(table)
+    pk_value = row[pks[0] if pks else "rowid"]
+    before = await db.execute_fn(lambda conn: list(conn.iterdump()))
+
+    if operation in ("create", "create_uppercase"):
+        path = "/data/-/create"
+        body = {
+            "table": table.upper() if operation == "create_uppercase" else table,
+            "rows": [row],
+        }
+    elif operation in ("update", "delete"):
+        path = f"/data/{table}/{pk_value}/-/{operation}"
+        body = {"update": row} if operation == "update" else {}
+    else:
+        path = f"/data/{table}/-/{operation}"
+        body = {"rows": [row]}
+    response = await ds_write.client.post(
+        path, json=body, headers=_headers(write_token(ds_write))
+    )
+    assert response.status_code == 400, response.text
+    assert response.json()["errors"] == ["Structured writes require an ordinary table"]
+    assert await db.execute_fn(lambda conn: list(conn.iterdump())) == before
+
+
+@pytest.mark.asyncio
+async def test_structured_writes_to_content_table_maintain_fts(ds_write):
+    db = ds_write.get_database("data")
+    await db.execute_write_fn(
+        lambda conn: sqlite_utils.Database(conn)["docs"].enable_fts(
+            ["title"], create_triggers=True
+        )
+    )
+    response = await ds_write.client.post(
+        "/data/docs/-/insert",
+        json={"row": {"id": 1, "title": "ordinary content"}},
+        headers=_headers(write_token(ds_write)),
+    )
+    assert response.status_code == 201, response.text
+    matches = await db.execute(
+        "select rowid from docs_fts where docs_fts match ?", ["ordinary"]
+    )
+    assert [row[0] for row in matches.rows] == [1]
+
+
 @pytest.mark.asyncio
 async def test_base64_write_api_create_table_infers_blob_and_raw_escapes(ds_write):
     token = write_token(ds_write)
@@ -1296,7 +1372,7 @@ async def test_alter_table_foreign_key_without_fk_column_requires_single_pk(ds_w
 
 @pytest.mark.asyncio
 async def test_foreign_key_suggestions(ds_write):
-    token = write_token(ds_write, permissions=["at"])
+    token = write_token(ds_write, permissions=["alter-table", "view-table"])
     db = ds_write.get_database("data")
     await db.execute_write("create table owners (id integer primary key)")
     await db.execute_write("insert into owners (id) values (1), (2), (3)")
@@ -1362,7 +1438,7 @@ async def test_foreign_key_suggestions_permission_denied(ds_write):
 
 @pytest.mark.asyncio
 async def test_foreign_key_suggestions_fail_open(ds_write, monkeypatch):
-    token = write_token(ds_write, permissions=["at"])
+    token = write_token(ds_write, permissions=["alter-table", "view-table"])
     db = ds_write.get_database("data")
     await db.execute_write("create table owners (id integer primary key)")
 
@@ -1393,7 +1469,7 @@ async def test_foreign_key_suggestions_fail_open(ds_write, monkeypatch):
 
 @pytest.mark.asyncio
 async def test_foreign_key_targets(ds_write):
-    token = write_token(ds_write, permissions=["ct"])
+    token = write_token(ds_write, permissions=["create-table", "view-table"])
     db = ds_write.get_database("data")
     await db.execute_write("create table owners (id integer primary key)")
     await db.execute_write("create table categories (slug varchar(30) primary key)")
@@ -2745,3 +2821,119 @@ async def test_create_using_alter_against_existing_table(
         insert_rows_event = ds_write._tracked_events[1]
         assert insert_rows_event.name == "insert-rows"
         assert insert_rows_event.num_rows == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("denied_action", "request_body"),
+    (
+        (
+            "insert-row",
+            {
+                "table": "salaries",
+                "rows": [{"id": 9, "note": "INJ-VIA-CREATE"}],
+            },
+        ),
+        (
+            "update-row",
+            {
+                "table": "salaries",
+                "rows": [{"id": 1, "note": "REPLACED"}],
+                "pk": "id",
+                "replace": True,
+            },
+        ),
+        (
+            "alter-table",
+            {
+                "table": "salaries",
+                "rows": [{"id": 9, "note": "INSERTED", "extra": "NEW"}],
+                "alter": True,
+            },
+        ),
+    ),
+)
+async def test_create_table_existing_table_respects_table_level_denial(
+    denied_action, request_body
+):
+    # GHSA-53fc-rhfg-h7qp issue 2: POST /db/-/create against an existing table
+    # inserts rows into it, so insert-row (and update-row / alter-table) must be
+    # checked against the TableResource, not just the DatabaseResource.
+    ds = Datasette(
+        memory=True,
+        config={
+            "databases": {
+                # id=editor user has each permission at the database level, but
+                # the selected action is explicitly denied on the salaries table
+                "data": {
+                    "permissions": {
+                        "create-table": {"id": "editor"},
+                        "insert-row": {"id": "editor"},
+                        "update-row": {"id": "editor"},
+                        "alter-table": {"id": "editor"},
+                    },
+                    "tables": {
+                        "salaries": {"permissions": {denied_action: False}},
+                    },
+                }
+            }
+        },
+    )
+    db = ds.add_memory_database(
+        f"create_table_existing_table_denied_{denied_action}", name="data"
+    )
+    await db.execute_write("create table salaries (id integer primary key, note text)")
+    await db.execute_write("insert into salaries values (1, 'TOPSECRET-A')")
+    await ds.invoke_startup()
+
+    if denied_action == "insert-row":
+        # Sanity: direct insert into salaries is denied for this actor
+        direct = await ds.client.post(
+            "/data/salaries/-/insert",
+            actor={"id": "editor"},
+            json={"row": {"id": 9, "note": "INJ-DIRECT"}},
+        )
+        assert direct.status_code == 403
+
+    response = await ds.client.post(
+        "/data/-/create",
+        actor={"id": "editor"},
+        json=request_body,
+    )
+    assert response.status_code == 403, response.json()
+    assert response.json()["errors"] == [f"Permission denied: need {denied_action}"]
+    rows = (await db.execute("select id, note from salaries order by id")).rows
+    assert [tuple(r) for r in rows] == [(1, "TOPSECRET-A")]
+    assert await db.table_columns("salaries") == ["id", "note"]
+
+
+@pytest.mark.asyncio
+async def test_create_table_respects_predeclared_table_level_denial():
+    ds = Datasette(
+        memory=True,
+        config={
+            "databases": {
+                "data": {
+                    "permissions": {
+                        "create-table": {"id": "editor"},
+                        "insert-row": {"id": "editor"},
+                    },
+                    "tables": {
+                        "planned_table": {"permissions": {"insert-row": False}},
+                    },
+                }
+            }
+        },
+    )
+    db = ds.add_memory_database("create_table_predeclared_denial", name="data")
+    await ds.invoke_startup()
+
+    response = await ds.client.post(
+        "/data/-/create",
+        actor={"id": "editor"},
+        json={"table": "planned_table", "rows": [{"id": 1}]},
+    )
+
+    assert response.status_code == 403, response.json()
+    assert response.json()["errors"] == ["Permission denied: need insert-row"]
+    assert not await db.table_exists("planned_table")

tests/test_auth.py

@@ -1,4 +1,5 @@
 import time
+from unittest.mock import AsyncMock
 
 import pytest
 from bs4 import BeautifulSoup as Soup
@@ -237,6 +238,35 @@ def test_auth_create_token(
         assert response3.json["actor"]["id"] == "test"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("method", ["GET", "POST"])
+@pytest.mark.parametrize(
+    "restrictions",
+    [
+        {},
+        {"a": ["vi"]},
+        {"d": {"db": ["vd"]}},
+        {"r": {"db": {"t1": ["vt"]}}},
+    ],
+    ids=["empty", "instance", "database", "table"],
+)
+async def test_auth_create_token_not_allowed_for_restricted_actors(
+    bare_ds, monkeypatch, method, restrictions
+):
+    create_token = AsyncMock()
+    monkeypatch.setattr(bare_ds, "create_token", create_token)
+
+    response = await bare_ds.client.request(
+        method,
+        "/-/create-token",
+        actor={"id": "test", "_r": restrictions},
+    )
+
+    assert response.status_code == 403
+    assert "Restricted actors cannot create API tokens" in response.text
+    create_token.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_auth_create_token_not_allowed_for_tokens(ds_client):
     ds_tok = ds_client.ds.sign(
@@ -524,3 +554,25 @@ async def test_root_without_root_enabled_no_special_permissions(ds_client):
         )
         is not True
     ), "Root without root_enabled should not automatically get set-column-type"
+
+
+@pytest.mark.parametrize("expire_after", (1, 300, 3600, 30 * 24 * 60 * 60))
+def test_set_actor_cookie_honours_expire_after(expire_after):
+    # GHSA-53fc-rhfg-h7qp issue 4: expire_after is documented as a number of
+    # seconds, but every value was being replaced with 24 hours.
+    from datasette.app import Datasette
+    from datasette.utils.asgi import Response
+
+    ds = Datasette(memory=True)
+    response = Response.text("")
+    before = int(time.time())
+    ds.set_actor_cookie(response, {"id": "test"}, expire_after=expire_after)
+    after = int(time.time())
+
+    (header,) = response._set_cookie_headers
+    assert header.startswith("ds_actor=")
+    value = header[len("ds_actor=") :].split(";", 1)[0]
+    data = ds.unsign(value, "actor")
+    assert data["a"] == {"id": "test"}
+    expires_at = baseconv.base62.decode(data["e"])
+    assert before + expire_after <= expires_at <= after + expire_after

tests/test_fts_permissions.py

@@ -0,0 +1,412 @@
+import pytest
+
+from datasette import hookimpl
+from datasette.app import Datasette
+from datasette.permissions import Action, PermissionSQL, _permission_check_cache
+from datasette.resources import DatabaseResource, TableResource
+from datasette.utils.sqlite import sqlite3, sqlite_derived_table_dependencies
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("fts_module", ["fts4", "fts5"])
+@pytest.mark.parametrize("actor", [None, {"id": "root"}], ids=["anonymous", "root"])
+async def test_derived_permissions_allow_one_hop_but_deny_nested_sources(
+    fts_module, actor
+):
+    class InspectPlugin:
+        @hookimpl
+        def register_actions(self):
+            return [
+                Action(
+                    name="inspect-derived",
+                    description="Inspect a table",
+                    resource_class=TableResource,
+                    also_requires="view-table",
+                )
+            ]
+
+        @hookimpl
+        def permission_resources_sql(self, action):
+            if action == "inspect-derived":
+                return PermissionSQL(
+                    sql="SELECT NULL AS parent, NULL AS child, 1 AS allow, 'inspect allowed' AS reason"
+                )
+
+    ds = Datasette(memory=True)
+    ds.pm.register(InspectPlugin(), name="inspect-derived-test")
+    db = ds.add_memory_database(
+        f"derived_one_hop_{fts_module}_{actor is not None}", name="data"
+    )
+    await db.execute_write("create table Documents (body text)")
+    await db.execute_write(
+        f"create virtual table Search using {fts_module}(body, content='Documents')"
+    )
+    await db.execute_write(
+        f"create virtual table Nested using {fts_module}(body, content='sEaRcH')"
+    )
+    await ds.invoke_startup()
+    token = _permission_check_cache.set({})
+    try:
+        # Both direct permissions are allowed, but a derived source makes its
+        # dependent unavailable even to an actor who can view the whole chain.
+        # Check and cache Search first so its cached grant cannot grant Nested.
+        for table, expected in (
+            ("Documents", True),
+            ("Search", True),
+            ("Nested", False),
+            ("Search_docsize", False),
+        ):
+            for spelling in (table, table.upper(), table.lower()):
+                assert await ds.allowed_many(
+                    actions=["view-table", "inspect-derived"],
+                    resource=TableResource("data", spelling),
+                    actor=actor,
+                ) == {"view-table": expected, "inspect-derived": expected}
+
+        page = await ds.allowed_resources(
+            "view-table", actor, parent="data", include_is_private=True, limit=1000
+        )
+        allowed = {resource.child for resource in page.resources}
+        assert {"Documents", "Search"}.issubset(allowed)
+        assert "Nested" not in allowed
+        assert "Search_docsize" not in allowed
+    finally:
+        _permission_check_cache.reset(token)
+        ds.pm.unregister(name="inspect-derived-test")
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("listing", [False, True], ids=["individual", "listing"])
+async def test_derived_permission_discovery_error_is_retried(monkeypatch, listing):
+    ds = Datasette(memory=True)
+    db = ds.add_memory_database(f"derived_discovery_error_{listing}", name="data")
+    await db.execute_write("create table documents (id integer primary key)")
+    await ds.invoke_startup()
+
+    class UnavailableSchema:
+        def execute(self, sql):
+            raise sqlite3.DatabaseError("schema temporarily unavailable")
+
+    async def check():
+        if listing:
+            return await ds.allowed_resources("view-table", parent="data")
+        return await ds.allowed(
+            action="view-table", resource=TableResource("data", "documents")
+        )
+
+    token = _permission_check_cache.set({})
+    try:
+        with monkeypatch.context() as patch:
+            patch.setattr(
+                "datasette.database.sqlite_derived_table_dependencies",
+                lambda conn: sqlite_derived_table_dependencies(UnavailableSchema()),
+            )
+            with pytest.raises(sqlite3.DatabaseError, match="schema temporarily"):
+                await check()
+
+        # Failed discovery must not cache an empty map or a permission grant.
+        assert db._cached_derived_table_dependencies is None
+        assert not _permission_check_cache.get()
+        result = await check()
+        if listing:
+            assert [resource.child for resource in result.resources] == ["documents"]
+        else:
+            assert result is True
+        assert db._cached_derived_table_dependencies is not None
+    finally:
+        _permission_check_cache.reset(token)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("fts_module", ("fts4", "fts5"))
+async def test_external_content_fts_inherits_content_table_view_permission(fts_module):
+    actor = {"id": "reader"}
+    secret_marker = "ISSUE_17_EXTERNAL_CONTENT_FTS_SECRET"
+    ds = Datasette(
+        memory=True,
+        config={
+            "permissions": {
+                "view-instance": {"id": "reader"},
+                "view-database": {"id": "reader"},
+                "view-table": {"id": "reader"},
+                "execute-sql": {"id": "nobody"},
+            },
+            "databases": {
+                "data": {
+                    "tables": {
+                        "secret": {"permissions": {"view-table": False}},
+                    }
+                }
+            },
+        },
+    )
+    db = ds.add_memory_database(f"issue_17_{fts_module}_permissions", name="data")
+    await db.execute_write("create table secret (id integer primary key, body text)")
+    await db.execute_write(
+        "insert into secret (body) values (?)",
+        [secret_marker],
+    )
+    fts_options = "body, content='secret'"
+    if fts_module == "fts5":
+        fts_options += ", content_rowid='id'"
+    await db.execute_write(
+        f"create virtual table secret_fts using {fts_module}({fts_options})"
+    )
+    await db.execute_write("insert into secret_fts(secret_fts) values ('rebuild')")
+    await ds.invoke_startup()
+
+    try:
+        assert "secret_fts" in await db.hidden_table_names()
+        assert (
+            await ds.allowed(
+                action="execute-sql",
+                resource=DatabaseResource("data"),
+                actor=actor,
+            )
+            is False
+        )
+
+        direct = await ds.client.get("/data/secret.json", actor=actor)
+        assert direct.status_code == 403
+
+        companion = await ds.client.get(
+            "/data/secret_fts.json?_shape=array",
+            actor=actor,
+        )
+        assert companion.status_code in (403, 404), (
+            "An automatically hidden external-content FTS table must inherit "
+            "the content table's view denial or be unavailable: "
+            f"{companion.text}"
+        )
+        assert secret_marker not in companion.text
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("fts_module", ("fts4", "fts5"))
+@pytest.mark.parametrize("contentless", (False, True), ids=("internal", "contentless"))
+async def test_fts_shadow_tables_inherit_logical_table_view_permission(
+    fts_module, contentless
+):
+    table_config = {
+        "secret_fts": {"permissions": {"view-table": False}},
+        # An explicit allow on one implementation table must not override
+        # the logical FTS table's denial.
+        "secret_fts_docsize": {"permissions": {"view-table": True}},
+    }
+    ds = Datasette(
+        memory=True,
+        config={
+            "permissions": {
+                "view-instance": True,
+                "view-database": True,
+                "view-table": True,
+                "execute-sql": False,
+            },
+            "databases": {"data": {"tables": table_config}},
+        },
+    )
+    db = ds.add_memory_database(
+        f"issue_17_{fts_module}_{'contentless' if contentless else 'internal'}",
+        name="data",
+    )
+    options = "body, content=''" if contentless else "body"
+    await db.execute_write(
+        f"create virtual table secret_fts using {fts_module}({options})"
+    )
+    await db.execute_write(
+        "insert into secret_fts(rowid, body) values (1, 'ISSUE_17_SHADOW_SECRET')"
+    )
+    await ds.invoke_startup()
+
+    try:
+        dependencies = await db.derived_table_dependencies()
+        shadow_tables = sorted(
+            table for table, source in dependencies.items() if source == "secret_fts"
+        )
+        assert shadow_tables
+        assert "secret_fts_docsize" in shadow_tables
+
+        for shadow_table in shadow_tables:
+            assert (
+                await ds.allowed(
+                    action="view-table",
+                    resource=TableResource("data", shadow_table),
+                )
+                is False
+            )
+            response = await ds.client.get(f"/data/{shadow_table}.json?_shape=array")
+            assert response.status_code == 403
+            assert "ISSUE_17_SHADOW_SECRET" not in response.text
+
+        allowed = await ds.allowed_resources("view-table", parent="data", limit=1000)
+        allowed_names = {resource.child for resource in allowed.resources}
+        assert not set(shadow_tables).intersection(allowed_names)
+
+        database_json = await ds.client.get("/data.json")
+        assert database_json.status_code == 200
+        for shadow_table in shadow_tables:
+            assert shadow_table not in database_json.text
+
+        schema_json = await ds.client.get("/data/-/schema.json")
+        assert schema_json.status_code == 200
+        for shadow_table in shadow_tables:
+            assert shadow_table not in schema_json.text
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "content_allowed,companion_allowed,expected",
+    (
+        (False, True, False),
+        (True, False, False),
+        (True, True, True),
+    ),
+)
+async def test_external_content_and_companion_permissions_are_both_required(
+    content_allowed, companion_allowed, expected
+):
+    ds = Datasette(
+        memory=True,
+        default_deny=True,
+        config={
+            "permissions": {
+                "view-instance": True,
+                "view-database": True,
+            },
+            "databases": {
+                "data": {
+                    "tables": {
+                        "secret": {"permissions": {"view-table": content_allowed}},
+                        "secret_fts": {
+                            "permissions": {"view-table": companion_allowed}
+                        },
+                    }
+                }
+            },
+        },
+    )
+    db = ds.add_memory_database(
+        f"issue_17_explicit_{int(content_allowed)}_{int(companion_allowed)}",
+        name="data",
+    )
+    await db.execute_write("create table secret(id integer primary key, body text)")
+    await db.execute_write("insert into secret(body) values ('ISSUE_17_MATRIX_SECRET')")
+    await db.execute_write(
+        "create virtual table secret_fts using fts5("
+        "body, content='secret', content_rowid='id')"
+    )
+    await db.execute_write("insert into secret_fts(secret_fts) values ('rebuild')")
+    await ds.invoke_startup()
+
+    try:
+        assert (
+            await ds.allowed(
+                action="view-table",
+                resource=TableResource("data", "secret_fts"),
+            )
+            is expected
+        )
+        response = await ds.client.get("/data/secret_fts.json?_shape=array")
+        assert response.status_code == (200 if expected else 403)
+        if not expected:
+            assert "ISSUE_17_MATRIX_SECRET" not in response.text
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_derived_tables_propagate_private_flag_and_route_permissions():
+    actor = {"id": "reader"}
+    ds = Datasette(
+        memory=True,
+        config={
+            "permissions": {
+                "view-instance": True,
+                "view-database": True,
+                "view-table": True,
+            },
+            "databases": {
+                "data": {
+                    "tables": {
+                        "secret": {"permissions": {"view-table": {"id": "reader"}}}
+                    }
+                }
+            },
+        },
+    )
+    db = ds.add_memory_database("issue_17_private_flag", name="data")
+    await db.execute_write("create table secret(id integer primary key, body text)")
+    await db.execute_write("insert into secret(body) values ('PRIVATE')")
+    await db.execute_write(
+        "create virtual table secret_fts using fts5("
+        "body, content='secret', content_rowid='id')"
+    )
+    await db.execute_write("insert into secret_fts(secret_fts) values ('rebuild')")
+    await ds.invoke_startup()
+
+    try:
+        actor_page = await ds.allowed_resources(
+            "view-table", actor, parent="data", include_is_private=True, limit=1000
+        )
+        actor_resources = {
+            resource.child: resource for resource in actor_page.resources
+        }
+        derived_names = set(await db.derived_table_dependencies())
+        assert "secret_fts" in actor_resources
+        assert actor_resources["secret_fts"].private
+        # Shadow tables depend on the already-derived external-content FTS
+        # table, so they remain unavailable even to the permitted reader.
+        assert not (derived_names - {"secret_fts"}).intersection(actor_resources)
+
+        anonymous_page = await ds.allowed_resources(
+            "view-table", parent="data", limit=1000
+        )
+        anonymous_names = {resource.child for resource in anonymous_page.resources}
+        assert not derived_names.intersection(anonymous_names)
+
+        for path in (
+            "/data/secret_fts.json?_facet=body",
+            "/data/secret_fts.csv",
+            "/data/secret_fts/-/autocomplete?q=PRIVATE",
+            "/data/secret_fts/-/schema.json",
+        ):
+            denied = await ds.client.get(path)
+            assert denied.status_code == 403
+            allowed = await ds.client.get(path, actor=actor)
+            assert allowed.status_code == 200
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_cyclic_derived_table_dependencies_fail_closed():
+    ds = Datasette(memory=True)
+    db = ds.add_memory_database("issue_17_cycle", name="data")
+    await db.execute_write(
+        "create virtual table first_fts using fts5(body, content='second_fts')"
+    )
+    await db.execute_write(
+        "create virtual table second_fts using fts5(body, content='first_fts')"
+    )
+    await ds.invoke_startup()
+
+    try:
+        for table in ("first_fts", "second_fts"):
+            assert (
+                await ds.allowed(
+                    action="view-table", resource=TableResource("data", table)
+                )
+                is False
+            )
+
+        page = await ds.allowed_resources("view-table", parent="data", limit=1000)
+        allowed_names = {resource.child for resource in page.resources}
+        assert "first_fts" not in allowed_names
+        assert "second_fts" not in allowed_names
+    finally:
+        ds.close()

tests/test_html.py

@@ -36,8 +36,10 @@ def test_homepage(app_client_two_attached_databases):
     h2 = soup.select("h2")[0]
     assert "extra database" == h2.text.strip()
     counts_p, links_p = h2.find_all_next("p")[:2]
+    # Shadow tables of the external-content index are denied, so they do not
+    # contribute to the table or row totals.
     assert (
-        "2 rows in 1 table, 5 rows in 4 hidden tables, 1 view" == counts_p.text.strip()
+        "2 rows in 1 table, 2 rows in 1 hidden table, 1 view" == counts_p.text.strip()
     )
     # We should only show visible, not hidden tables here:
     table_links = [

tests/test_internals_database.py

@@ -15,11 +15,16 @@ from datasette.database import (
     DatasetteClosedError,
     ExecuteWriteResult,
     MultipleValues,
+    QueryInterrupted,
     Results,
     _deliver_write_result,
 )
 from datasette.utils import Column
-from datasette.utils.sqlite import sqlite3, supports_returning
+from datasette.utils.sqlite import (
+    sqlite3,
+    sqlite_derived_table_dependencies,
+    supports_returning,
+)
 
 requires_sqlite_returning = pytest.mark.skipif(
     not supports_returning(), reason="SQLite does not support RETURNING"
@@ -38,6 +43,31 @@ async def test_execute(db):
     assert 15 == len(results)
 
 
+@pytest.mark.asyncio
+async def test_derived_dependency_cache_survives_failed_refresh(monkeypatch):
+    ds = Datasette(memory=True)
+    db = ds.add_memory_database(uuid.uuid4().hex, name="data")
+    await db.derived_table_dependencies()
+    previous_cache = db._cached_derived_table_dependencies
+    await db.execute_write("create table dependency_cache_refresh (id integer)")
+
+    class UnavailableSchema:
+        def execute(self, sql):
+            raise sqlite3.DatabaseError("schema temporarily unavailable")
+
+    with monkeypatch.context() as patch:
+        patch.setattr(
+            "datasette.database.sqlite_derived_table_dependencies",
+            lambda conn: sqlite_derived_table_dependencies(UnavailableSchema()),
+        )
+        with pytest.raises(sqlite3.DatabaseError, match="schema temporarily"):
+            await db.derived_table_dependencies()
+    assert db._cached_derived_table_dependencies == previous_cache
+
+    await db.derived_table_dependencies()
+    assert db._cached_derived_table_dependencies[0] != previous_cache[0]
+
+
 @pytest.mark.asyncio
 async def test_results_first(db):
     assert None is (await db.execute("select * from facetable where pk > 100")).first()
@@ -478,6 +508,31 @@ async def test_view_names(db):
     ]
 
 
+@pytest.mark.asyncio
+async def test_execute_write_custom_time_limit():
+    ds = Datasette(settings={"sql_time_limit_ms": 1})
+    db = ds.add_memory_database(uuid.uuid4().hex, name="write_limits")
+    await ds.invoke_startup()
+    # Bounded work from PR #51; even without a limit this finishes on its own.
+    sql = (
+        "with recursive c(x) as "
+        "(select 1 union all select x+1 from c where x < 800000) "
+        "select x from c where x < 0"
+    )
+    try:
+        await db.execute_write("create table items(value integer)")
+        with pytest.raises(QueryInterrupted):
+            await db.execute(sql)
+        # Writes take their own explicit limit, independent of the read setting.
+        with pytest.raises(QueryInterrupted):
+            await db.execute_write(f"insert into items(value) {sql}", time_limit_ms=1)
+        # Interruption must leave the connection available for subsequent writes.
+        await db.execute_write("insert into items(value) values (1)")
+        assert (await db.execute("select value from items")).single_value() == 1
+    finally:
+        ds.close()
+
+
 @pytest.mark.asyncio
 async def test_execute_write_block_true(db):
     result = await db.execute_write(
@@ -705,6 +760,33 @@ async def test_execute_write_fn_block_false(db):
     assert isinstance(task_id, uuid.UUID)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("disable_threads", (False, True))
+async def test_execute_write_fn_block_false_returns_uuid(tmp_path, disable_threads):
+    # block=False is documented to return "a UUID representing the queued task".
+    # With num_sql_threads=0 there is no write thread, so the non-threaded branch
+    # has to satisfy the same contract as the threaded one.
+    settings = {"num_sql_threads": 0} if disable_threads else {}
+    ds = Datasette([], memory=True, settings=settings)
+    await ds.invoke_startup()
+    db = ds.add_memory_database("test_block_false")
+    await db.execute_write(
+        "create table if not exists t (id integer primary key, v text)"
+    )
+
+    def write_fn(conn):
+        conn.execute("insert into t (v) values ('a')")
+        # Returns None, like most write functions.
+
+    task_id = await db.execute_write_fn(write_fn, block=False)
+
+    assert isinstance(task_id, uuid.UUID)
+    # Distinct per call, so a caller can tell two queued tasks apart.
+    second = await db.execute_write_fn(write_fn, block=False)
+    assert isinstance(second, uuid.UUID)
+    assert second != task_id
+
+
 @pytest.mark.asyncio
 async def test_execute_write_fn_block_true(db):
     def write_fn(conn):

tests/test_load_extensions.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+from unittest import mock
 
 import pytest
 
@@ -20,6 +21,29 @@ def has_compiled_ext():
     return False
 
 
+@pytest.mark.parametrize("load_fails", (False, True))
+def test_load_extension_is_disabled(load_fails):
+    ds = Datasette(sqlite_extensions=[COMPILED_EXTENSION_PATH])
+    connection = mock.Mock()
+    if load_fails:
+        connection.load_extension.side_effect = RuntimeError
+
+    if load_fails:
+        with pytest.raises(RuntimeError):
+            ds._prepare_connection(connection, "data")
+    else:
+        ds._prepare_connection(connection, "data")
+
+    # Extensions are loaded using the Python API, never via SQL
+    assert connection.load_extension.mock_calls == [
+        mock.call(COMPILED_EXTENSION_PATH),
+    ]
+    assert connection.enable_load_extension.mock_calls == [
+        mock.call(True),
+        mock.call(False),
+    ]
+
+
 @pytest.mark.asyncio
 @pytest.mark.skipif(not has_compiled_ext(), reason="Requires compiled ext.c")
 async def test_load_extension_default_entrypoint():
@@ -64,3 +88,20 @@ async def test_load_extension_multiple_entrypoints():
     response = await ds.client.get("/_memory/-/query.json?_shape=arrays&sql=select+c()")
     assert response.status_code == 200
     assert response.json()["rows"][0][0] == "c"
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(not has_compiled_ext(), reason="Requires compiled ext.c")
+async def test_sql_cannot_load_additional_extension():
+    ds = Datasette(sqlite_extensions=[COMPILED_EXTENSION_PATH])
+
+    response = await ds.client.get(
+        "/_memory/-/query.json",
+        params={
+            "sql": "select load_extension(:path, :entrypoint)",
+            "path": COMPILED_EXTENSION_PATH,
+            "entrypoint": "sqlite3_ext_b_init",
+        },
+    )
+    assert response.status_code == 400
+    assert response.json()["error"] == "not authorized"

tests/test_permission_endpoints.py

@@ -494,3 +494,31 @@ async def test_execute_sql_requires_view_database():
         )
     finally:
         ds.pm.unregister(plugin)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("path", ["/-/allowed", "/-/allowed.json?action=view-table"])
+async def test_allowed_requires_view_instance(path):
+    """
+    GHSA-hp2x-vx2r-6vxg: /-/allowed should be gated like its /-/rules sibling.
+
+    An actor who is denied view-instance gets 403 from / and /-/rules, but
+    /-/allowed (HTML and JSON) currently returns 200 to the same actor.
+    """
+    ds = Datasette(config={"allow": {"id": "alice"}})
+    await ds.invoke_startup()
+    db = ds.add_memory_database("live")
+    await db.execute_write("CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY)")
+    await ds.refresh_schemas()
+
+    assert (await ds.client.get("/")).status_code == 403
+    assert (await ds.client.get("/-/rules.json?action=view-table")).status_code == 403
+
+    response = await ds.client.get(path)
+    assert response.status_code == 403
+
+    # Alice is still allowed
+    response = await ds.client.get(
+        path, cookies={"ds_actor": ds.client.actor_cookie({"id": "alice"})}
+    )
+    assert response.status_code == 200

tests/test_pr76_fts_policy.py

@@ -0,0 +1,140 @@
+"""Policy and compatibility coverage for PR #76, run against the fixed checkout."""
+
+import uuid
+
+import pytest
+
+from datasette.app import Datasette
+from datasette.resources import TableResource
+from datasette.utils.sqlite import sqlite3, sqlite_derived_table_dependencies
+
+
+@pytest.mark.parametrize("vocab_name", ["words", "name USING fts4aux", 'quoted"name'])
+@pytest.mark.parametrize(
+    "module,arguments",
+    [
+        ("fts5vocab", "'Search,Index', 'row'"),
+        ("fts5vocab", "'SEARCH,INDEX', 'col'"),
+        ("fts5vocab", "'Search,Index', 'instance'"),
+        ("fts4aux", "'Search,Index'"),
+    ],
+)
+def test_vocabulary_dependency_identity(module, arguments, vocab_name):
+    conn = sqlite3.connect(":memory:")
+    try:
+        fts = "fts5" if module == "fts5vocab" else "fts4"
+        conn.execute(f'create virtual table "Search,Index" using {fts}(body)')
+        quoted_name = '"' + vocab_name.replace('"', '""') + '"'
+        conn.execute(
+            f"create virtual table {quoted_name} USING /* module */ {module}({arguments})"
+        )
+        assert sqlite_derived_table_dependencies(conn)[vocab_name] == "Search,Index"
+    finally:
+        conn.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("module", ["fts5", "fts4"])
+@pytest.mark.parametrize("external_content", [False, True], ids=["one-hop", "two-hop"])
+@pytest.mark.parametrize(
+    "source_allowed,vocab_allowed", [(False, True), (True, False), (True, True)]
+)
+async def test_vocabulary_immediate_source_permissions(
+    module, external_content, source_allowed, vocab_allowed
+):
+    ds = Datasette(
+        memory=True,
+        config={
+            "databases": {
+                "data": {
+                    "tables": {
+                        "search": {
+                            "permissions": {
+                                "view-table": (
+                                    {"id": "reader"} if source_allowed else False
+                                )
+                            }
+                        },
+                        "words": {"permissions": {"view-table": vocab_allowed}},
+                    }
+                }
+            }
+        },
+    )
+    db = ds.add_memory_database(uuid.uuid4().hex, name="data")
+    await db.execute_write("create table documents(body text)")
+    options = "body, content='documents'" if external_content else "body"
+    await db.execute_write(f"create virtual table search using {module}({options})")
+    definition = (
+        "fts5vocab('SEARCH', 'row')" if module == "fts5" else "fts4aux('SEARCH')"
+    )
+    await db.execute_write(f"create virtual table words using {definition}")
+    await ds.invoke_startup()
+    try:
+        actor = {"id": "reader"}
+        expected = source_allowed and vocab_allowed and not external_content
+        for name in ("words", "WORDS"):
+            assert (
+                await ds.allowed(
+                    action="view-table",
+                    resource=TableResource("data", name),
+                    actor=actor,
+                )
+                is expected
+            )
+        resources = await ds.allowed_resources(
+            "view-table", parent="data", actor=actor, include_is_private=True
+        )
+        words = [r for r in resources.resources if r.child == "words"]
+        assert bool(words) is expected
+        if expected:
+            assert words[0].private
+        assert not await ds.allowed(
+            action="view-table", resource=TableResource("data", "words")
+        )
+        # Dropping the source invalidates dependency metadata and remains denied.
+        await db.execute_write("drop table search")
+        assert not await ds.allowed(
+            action="view-table", resource=TableResource("data", "words"), actor=actor
+        )
+    finally:
+        ds.close()
+
+
+@pytest.mark.parametrize(
+    "module,definition",
+    [
+        ("fts5", "fts5vocab('main', 'search', 'row')"),
+        ("fts4", "fts4aux('main', 'search')"),
+    ],
+)
+def test_cross_schema_vocabulary_is_unresolved(module, definition):
+    conn = sqlite3.connect(":memory:")
+    try:
+        conn.execute(f"create virtual table search using {module}(body)")
+        conn.execute(f"create virtual table temp.words using {definition}")
+        # Cross-schema ownership is not representable by the current map.
+        # The source is itself derived, so the immediate-source policy denies it.
+        assert (
+            sqlite_derived_table_dependencies(conn, schema="temp")["words"] == "words"
+        )
+    finally:
+        conn.close()
+
+
+@pytest.mark.parametrize(
+    "definition",
+    [
+        """CREATE VIRTUAL TABLE"words"USING"fts5vocab"('search', 'row')""",
+        """CREATE VIRTUAL TABLE[words]USING[fts5vocab]('search', 'row')""",
+        """CREATE VIRTUAL TABLE`words`USING`fts5vocab`('search', 'row')""",
+    ],
+)
+def test_vocabulary_quoted_token_boundaries(definition):
+    conn = sqlite3.connect(":memory:")
+    try:
+        conn.execute("create virtual table search using fts5(body)")
+        conn.execute(definition)
+        assert sqlite_derived_table_dependencies(conn)["words"] == "search"
+    finally:
+        conn.close()

tests/test_pr76_statistics_policy.py

@@ -0,0 +1,113 @@
+"""Statistics access policy and plugin replacement coverage for PR #76."""
+
+import uuid
+
+import pytest
+
+from datasette import hookimpl
+from datasette.app import Datasette
+from datasette.permissions import PermissionSQL
+from datasette.resources import TableResource
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("scope", [None, "global", "database", "table", "root"])
+async def test_statistics_denied_despite_allow_rules(scope):
+    config = {"databases": {"data": {"tables": {"sqlite_stat1": {}}}}}
+    grant = {"view-table": True}
+    if scope == "global":
+        config["permissions"] = grant
+    elif scope == "database":
+        config["databases"]["data"]["permissions"] = grant
+    elif scope == "table":
+        config["databases"]["data"]["tables"]["sqlite_stat1"]["permissions"] = grant
+    ds = Datasette(memory=True, config=config)
+    ds.root_enabled = scope == "root"
+    actor = {"id": "root"} if scope == "root" else {"id": "reader"}
+    db = ds.add_memory_database(uuid.uuid4().hex, name="data")
+    await db.execute_write("create table items(value text)")
+    await db.execute_write("create index items_value on items(value)")
+    await db.execute_write("insert into items values ('example')")
+    await db.execute_write("analyze")
+    await ds.invoke_startup()
+    try:
+        assert "view-sqlite-statistics" not in ds.actions
+        for name in ("sqlite_stat1", "SQLITE_STAT1"):
+            assert not await ds.allowed(
+                action="view-table", resource=TableResource("data", name), actor=actor
+            )
+        for suffix in ("", ".json", ".csv"):
+            assert (
+                await ds.client.get(f"/data/sqlite_stat1{suffix}", actor=actor)
+            ).status_code == 403
+        resources = await ds.allowed_resources("view-table", parent="data", actor=actor)
+        assert "sqlite_stat1" not in {r.child for r in resources.resources}
+        assert "items" in {r.child for r in resources.resources}
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "table", ["sqlite_stat1", "sqlite_stat2", "sqlite_stat3", "sqlite_stat4"]
+)
+@pytest.mark.parametrize("default_deny", [False, True])
+async def test_statistics_names_denied(table, default_deny):
+    ds = Datasette(memory=True, default_deny=default_deny)
+    ds.root_enabled = True
+    await ds.invoke_startup()
+    try:
+        for name in (table, table.upper()):
+            assert not await ds.allowed(
+                action="view-table",
+                resource=TableResource("_memory", name),
+                actor={"id": "root"},
+            )
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_plugin_can_replace_statistics_policy():
+    class ReplacementPolicy:
+        @hookimpl
+        def permission_resources_sql(self, action, actor):
+            if action == "view-table":
+                return PermissionSQL(
+                    sql="SELECT 'data' AS parent, 'sqlite_stat1' AS child, :statistics_allowed AS allow, 'custom statistics policy' AS reason",
+                    params={"statistics_allowed": int(actor == {"id": "reader"})},
+                )
+
+    ds = Datasette(memory=True)
+    db = ds.add_memory_database(uuid.uuid4().hex, name="data")
+    await db.execute_write("create table items(value text)")
+    await db.execute_write("analyze")
+    await ds.invoke_startup()
+    name = "datasette.default_permissions.sqlite_statistics"
+    original = ds.pm.unregister(name=name)
+    assert original is not None
+    replacement = ReplacementPolicy()
+    ds.pm.register(replacement, name="test-replacement-statistics-policy")
+    try:
+        actor = {"id": "reader"}
+        assert await ds.allowed(
+            action="view-table",
+            resource=TableResource("data", "sqlite_stat1"),
+            actor=actor,
+        )
+        assert not await ds.allowed(
+            action="view-table", resource=TableResource("data", "sqlite_stat1")
+        )
+        resources = await ds.allowed_resources(
+            "view-table", parent="data", actor=actor, include_is_private=True
+        )
+        stats = [r for r in resources.resources if r.child == "sqlite_stat1"]
+        assert len(stats) == 1 and stats[0].private
+        assert (
+            await ds.client.get("/data/sqlite_stat1.json", actor=actor)
+        ).status_code == 200
+        assert (await ds.client.get("/data/sqlite_stat1.json")).status_code == 403
+    finally:
+        ds.pm.unregister(replacement)
+        ds.pm.register(original, name=name)
+        ds.close()

tests/test_queries.py

@@ -3248,74 +3248,6 @@ async def test_execute_write_create_table_uses_create_table_permission():
     assert not await db.table_exists("should_not_exist")
 
 
-@pytest.mark.asyncio
-async def test_execute_write_create_view_uses_create_view_permission():
-    ds = Datasette(
-        memory=True,
-        default_deny=True,
-        config={
-            "permissions": {
-                "insert-row": {"id": "row-writer"},
-                "update-row": {"id": "row-writer"},
-            },
-            "databases": {
-                "data": {
-                    "permissions": {
-                        "view-database": {"id": ["creator", "row-writer"]},
-                        "execute-write-sql": {"id": ["creator", "row-writer"]},
-                        "create-view": {"id": "creator"},
-                    }
-                }
-            },
-        },
-    )
-    db = ds.add_memory_database("execute_write_create_view", name="data")
-    await db.execute_write("create table dogs (id integer primary key, name text)")
-    await ds.invoke_startup()
-
-    analysis_response = await ds.client.get(
-        "/data/-/execute-write/analyze",
-        actor={"id": "creator"},
-        params={"sql": "create view dog_names as select id, name from dogs"},
-    )
-    allowed_response = await ds.client.post(
-        "/data/-/execute-write",
-        actor={"id": "creator"},
-        json={"sql": "create view dog_names as select id, name from dogs"},
-    )
-    row_permission_response = await ds.client.post(
-        "/data/-/execute-write",
-        actor={"id": "row-writer"},
-        json={"sql": "create view should_not_exist as select id from dogs"},
-    )
-
-    assert analysis_response.status_code == 200
-    analysis_data = analysis_response.json()
-    assert analysis_data["ok"] is True
-    assert analysis_data["execute_disabled"] is False
-    assert analysis_data["analysis_rows"] == [
-        {
-            "operation": "create",
-            "database": "data",
-            "table": "dog_names",
-            "required_permission": "create-view",
-            "source": None,
-            "allowed": True,
-        }
-    ]
-
-    assert allowed_response.status_code == 200
-    assert allowed_response.json()["ok"] is True
-    assert allowed_response.json()["message"] == "Query executed"
-    assert await db.view_exists("dog_names")
-
-    assert row_permission_response.status_code == 403
-    assert row_permission_response.json()["errors"] == [
-        "Permission denied: need create-view on data"
-    ]
-    assert not await db.view_exists("should_not_exist")
-
-
 @pytest.mark.parametrize(
     (
         "database_name",

tests/test_schema_endpoints.py

@@ -246,3 +246,114 @@ async def test_table_not_exists(schema_ds):
     response = await schema_ds.client.get("/schema_public_db/nonexistent/-/schema.md")
     assert response.status_code == 404
     assert "not found" in response.text.lower()
+
+
+@pytest_asyncio.fixture(scope="module")
+async def schema_table_perms_ds():
+    """
+    A database that is viewable by anonymous users, but with one table
+    locked down using the documented per-table lockdown recipe:
+    a table-level allow block combined with allow_sql: false.
+    """
+    ds = Datasette(
+        config={
+            "databases": {
+                "schema_table_perms_db": {
+                    "allow_sql": False,
+                    "tables": {"employee_salaries": {"allow": {"id": "root"}}},
+                }
+            }
+        }
+    )
+    db = ds.add_memory_database("schema_table_perms_db")
+    await db.execute_write(
+        "CREATE TABLE IF NOT EXISTS public_posts (id INTEGER PRIMARY KEY, title TEXT)"
+    )
+    await db.execute_write(
+        "CREATE TABLE IF NOT EXISTS employee_salaries "
+        "(id INTEGER PRIMARY KEY, ssn TEXT, salary_usd INTEGER)"
+    )
+    await db.execute_write(
+        "CREATE INDEX IF NOT EXISTS idx_employee_salaries_ssn ON employee_salaries(ssn)"
+    )
+    await db.execute_write(
+        "CREATE TRIGGER IF NOT EXISTS trg_employee_salaries "
+        "AFTER INSERT ON employee_salaries BEGIN SELECT 1; END"
+    )
+    return ds
+
+
+@pytest.mark.asyncio
+async def test_schema_table_perms_controls(schema_table_perms_ds):
+    """Sanity check: the locked down table really is denied to anonymous users."""
+    ds = schema_table_perms_ds
+    for path in (
+        "/schema_table_perms_db/employee_salaries.json",
+        "/schema_table_perms_db/employee_salaries/-/schema.json",
+        "/schema_table_perms_db/-/query.json?sql=select+*+from+employee_salaries",
+    ):
+        response = await ds.client.get(path)
+        assert response.status_code == 403, path
+    response = await ds.client.get("/schema_table_perms_db.json")
+    assert response.status_code == 200
+    assert "employee_salaries" not in response.text
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "base_url",
+    ["/-/schema", "/schema_table_perms_db/-/schema"],
+)
+@pytest.mark.parametrize("format_ext", ["json", "md", ""])
+async def test_schema_parent_views_hide_denied_tables(
+    schema_table_perms_ds, base_url, format_ext
+):
+    """
+    GHSA-926p-cw2f-643h: /-/schema and /db/-/schema must not disclose the DDL
+    of tables the actor is denied view-table on, including indexes and
+    triggers that belong to those tables.
+    """
+    url = base_url + (f".{format_ext}" if format_ext else "")
+
+    # Anonymous: allowed table visible, denied table (and its columns,
+    # index and trigger) absent
+    response = await schema_table_perms_ds.client.get(url)
+    assert response.status_code == 200
+    assert "public_posts" in response.text
+    assert "employee_salaries" not in response.text
+    assert "ssn" not in response.text
+    assert "salary_usd" not in response.text
+    assert "idx_employee_salaries_ssn" not in response.text
+    assert "trg_employee_salaries" not in response.text
+
+    # root can see everything
+    response = await schema_table_perms_ds.client.get(url, actor={"id": "root"})
+    assert response.status_code == 200
+    assert "public_posts" in response.text
+    assert "CREATE TABLE employee_salaries" in response.text
+    assert "idx_employee_salaries_ssn" in response.text
+    assert "trg_employee_salaries" in response.text
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "object_name", ["idx_employee_salaries_ssn", "trg_employee_salaries"]
+)
+@pytest.mark.parametrize("format_ext", ["json", "md", ""])
+async def test_table_schema_does_not_serve_objects_of_denied_table(
+    schema_table_perms_ds, object_name, format_ext
+):
+    """
+    Related to GHSA-926p-cw2f-643h: /db/<name>/-/schema looks up sqlite_master
+    by name without restricting to tables/views, so requesting the name of an
+    index or trigger that belongs to a denied table serves its DDL. The
+    view-table check runs against the index/trigger name, which is not a
+    restricted table, so it passes.
+    """
+    url = f"/schema_table_perms_db/{object_name}/-/schema"
+    if format_ext:
+        url += f".{format_ext}"
+    response = await schema_table_perms_ds.client.get(url)
+    assert response.status_code in (403, 404)
+    assert "employee_salaries" not in response.text
+    assert "ssn" not in response.text

tests/test_stored_queries.py

@@ -208,11 +208,12 @@ def test_custom_params(stored_write_client):
     )
 
 
-def test_stored_query_pages_no_vary_header(stored_write_client):
-    # These pages no longer embed per-cookie CSRF tokens, so they must not
-    # set Vary: Cookie - they should be cacheable across users.
-    assert "vary" not in stored_write_client.get("/data").headers
-    assert "vary" not in stored_write_client.get("/data/update_name").headers
+def test_stored_query_pages_vary_by_credentials(stored_write_client):
+    # Even without per-cookie CSRF tokens, anonymous pages must not be reused
+    # for authenticated users whose permissions or navigation can differ.
+    for path in ("/data", "/data/update_name"):
+        response = stored_write_client.get(path)
+        assert response.headers["vary"] == "Cookie, Authorization"
 
 
 def test_json_post_body(stored_write_client):

tests/test_table_api.py

@@ -619,7 +619,10 @@ def test_searchmode(table_metadata, querystring, expected_rows):
             ],
         ),
         (
-            "/fixtures/searchable_view.json?_shape=arrays&_search=weasel&_fts_table=searchable_fts&_fts_pk=pk",
+            (
+                "/fixtures/searchable_view_configured_by_metadata.json"
+                "?_shape=arrays&_search=weasel&_fts_table=searchable_fts&_fts_pk=pk"
+            ),
             [[2, "terry dog", "sara weasel", "puma"]],
         ),
     ],
@@ -1778,3 +1781,34 @@ async def test_next_url_included_by_default(ds_client):
         data = response.json()
     assert data["next"] is None
     assert data["next_url"] is None
+
+
+@pytest.mark.asyncio
+async def test_table_through_requires_view_table_on_through_table():
+    # GHSA-53fc-rhfg-h7qp issue 3: ?_through= runs a sub-select against the
+    # caller-supplied through table, so the actor must be allowed to view it.
+    # Otherwise it is an equality oracle over any column of a denied table.
+    from datasette.app import Datasette
+
+    ds = Datasette(
+        memory=True,
+        config={"databases": {"data": {"tables": {"salaries": {"allow": False}}}}},
+    )
+    db = ds.add_memory_database("table_through_denied", name="data")
+    await db.execute_write("create table people (id integer primary key, name text)")
+    await db.execute_write(
+        "create table salaries (id integer primary key, "
+        "person_id integer references people(id), note text)"
+    )
+    await db.execute_write("insert into people values (1, 'alice'), (2, 'bob')")
+    await db.execute_write("insert into salaries values (1, 1, 'TOPSECRET-A')")
+    await ds.invoke_startup()
+
+    # Sanity: anonymous cannot read salaries directly
+    assert (await ds.client.get("/data/salaries.json")).status_code == 403
+
+    response = await ds.client.get(
+        "/data/people.json?_shape=array"
+        '&_through={"table":"salaries","column":"note","value":"TOPSECRET-A"}'
+    )
+    assert response.status_code == 403, response.text

tests/test_table_html.py

@@ -270,7 +270,8 @@ async def test_empty_search_parameter_gets_removed(ds_client):
 async def test_searchable_view_persists_fts_table(ds_client):
     # The search form should persist ?_fts_table as a hidden field
     response = await ds_client.get(
-        "/fixtures/searchable_view?_fts_table=searchable_fts&_fts_pk=pk"
+        "/fixtures/searchable_view_configured_by_metadata"
+        "?_fts_table=searchable_fts&_fts_pk=pk"
     )
     inputs = Soup(response.text, "html.parser").find("form").find_all("input")
     hiddens = [i for i in inputs if i["type"] == "hidden"]

tests/test_table_resource_identity.py

@@ -0,0 +1,420 @@
+"""Table permission identities must agree with SQLite identifier resolution."""
+
+import uuid
+from unittest.mock import AsyncMock
+
+import pytest
+
+from datasette import hookimpl
+from datasette.app import Datasette
+from datasette.default_permissions import restrictions_allow_action
+from datasette.permissions import Action, PermissionSQL, _permission_check_cache
+from datasette.resources import QueryResource, TableResource
+from datasette.utils.actions_sql import explain_permission_for_resource
+from datasette.utils.asgi import Forbidden
+from datasette.utils.permissions import gather_permission_sql_from_hooks
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["table", "view"])
+@pytest.mark.parametrize("spelling", ["Inventory", "inventory", "INVENTORY"])
+@pytest.mark.parametrize("allowed", [False, True])
+@pytest.mark.parametrize("rule_spelling", ["Inventory", "iNvEnToRy"])
+async def test_table_permission_identity(
+    kind, spelling, allowed, rule_spelling, monkeypatch
+):
+    ds = Datasette(
+        config={
+            "permissions": {"view-table": not allowed, "insert-row": not allowed},
+            "databases": {
+                "data": {
+                    "tables": {
+                        rule_spelling: {
+                            "permissions": {
+                                "view-table": allowed,
+                                "insert-row": allowed,
+                            }
+                        }
+                    }
+                }
+            },
+        }
+    )
+    db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data")
+    cache_token = _permission_check_cache.set({})
+    try:
+        await db.execute_write(
+            "create table Inventory (id integer primary key)"
+            if kind == "table"
+            else "create view Inventory as select 1 as id"
+        )
+        await ds.invoke_startup()
+        # Identity matching needs no target-schema lookup. Derived-table
+        # permissions may still check the schema version. All spellings and
+        # API entry points should share the existing permission result cache.
+        target_execute = AsyncMock(wraps=db.execute)
+        monkeypatch.setattr(db, "execute", target_execute)
+        internal_execute = AsyncMock(wraps=ds.get_internal_database().execute)
+        monkeypatch.setattr(ds.get_internal_database(), "execute", internal_execute)
+        resource = TableResource("data", spelling)
+        assert await ds.allowed_many(
+            actions=["view-table", "insert-row"], resource=resource
+        ) == {"view-table": allowed, "insert-row": allowed}
+        assert await ds.allowed(action="view-table", resource=resource) is allowed
+        assert await ds.check_visibility(None, "view-table", resource) == (
+            allowed,
+            False,
+        )
+        if allowed:
+            await ds.ensure_permission(action="view-table", resource=resource)
+        else:
+            with pytest.raises(Forbidden):
+                await ds.ensure_permission(action="view-table", resource=resource)
+        assert resource.child == spelling  # Do not mutate caller-owned resources.
+        for variant in ("Inventory", "inventory", "INVENTORY"):
+            assert (
+                await ds.allowed(
+                    action="view-table", resource=TableResource("data", variant)
+                )
+                is allowed
+            )
+        assert internal_execute.await_count == 1
+        assert all(
+            call.args[0] == "PRAGMA schema_version"
+            for call in target_execute.await_args_list
+        )
+        assert all(key[3] == "inventory" for key in _permission_check_cache.get())
+    finally:
+        _permission_check_cache.reset(cache_token)
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_other_permission_identities_are_preserved():
+    ds = Datasette(
+        config={
+            "databases": {
+                "data": {
+                    "tables": {
+                        "Äpfel": {"permissions": {"view-table": False}},
+                        "Future": {"permissions": {"view-table": False}},
+                    },
+                    "queries": {
+                        "Report": {
+                            "sql": "select 1",
+                            "permissions": {"view-query": False},
+                        },
+                        "report": {
+                            "sql": "select 1",
+                            "permissions": {"view-query": True},
+                        },
+                    },
+                }
+            }
+        }
+    )
+    db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data")
+    try:
+        await db.execute_write('create table "Äpfel" (id integer primary key)')
+        await db.execute_write('create table "äpfel" (id integer primary key)')
+        await db.execute_write("create table Report (id integer primary key)")
+        await ds.invoke_startup()
+        # SQLite folds ASCII identifier casing, not Unicode casing.
+        for name, expected in [
+            ("ÄPFEL", False),
+            ("äPFEL", True),
+            ("Future", False),
+            ("future", False),
+        ]:
+            assert (
+                await ds.allowed(
+                    action="view-table", resource=TableResource("data", name)
+                )
+                is expected
+            )
+        # Query names remain case-sensitive even when a table has the same name.
+        for name, expected in [("Report", False), ("report", True)]:
+            assert (
+                await ds.allowed(
+                    action="view-query", resource=QueryResource("data", name)
+                )
+                is expected
+            )
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("allow", [True, False, {"id": "reader"}])
+async def test_table_listings_and_explanations(allow):
+    ds = Datasette(
+        config={
+            "databases": {
+                "data": {
+                    "tables": {
+                        "inventory": {"permissions": {"view-table": allow}},
+                    }
+                }
+            }
+        }
+    )
+    db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data")
+    try:
+        await db.execute_write("create table Inventory (id integer primary key)")
+        await db.execute_write("create view InventoryView as select id from Inventory")
+        await ds.invoke_startup()
+        for actor in (None, {"id": "reader"}):
+            expected = allow is True or (isinstance(allow, dict) and actor == allow)
+            explanation = await explain_permission_for_resource(
+                datasette=ds,
+                actor=actor,
+                action="view-table",
+                parent="data",
+                child="INVENTORY",
+            )
+            assert explanation["allowed"] is expected
+            assert explanation["winning_scope"] == "resource"
+            assert any(
+                "data/inventory" in rule["reason"]
+                for rule in explanation["matched_rules"]
+            )
+            page = await ds.allowed_resources(
+                "view-table",
+                actor,
+                parent="data",
+                include_is_private=True,
+                include_reasons=True,
+                limit=1,
+            )
+            resources = [resource async for resource in page.all()]
+            matching = [r for r in resources if r.child == "Inventory"]
+            assert bool(matching) is expected
+            assert len(matching) <= 1
+            if matching:
+                assert matching[0].private is isinstance(allow, dict)
+            assert any(r.child == "InventoryView" for r in resources)
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("deny_first", [True, False])
+async def test_case_variant_rules_deny_wins(deny_first):
+    rules = [("inventory", False), ("INVENTORY", True)]
+    if not deny_first:
+        rules.reverse()
+    ds = Datasette(
+        config={
+            "databases": {
+                "data": {
+                    "tables": {
+                        name: {"permissions": {"view-table": allow}}
+                        for name, allow in rules
+                    }
+                }
+            }
+        }
+    )
+    db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data")
+    try:
+        await db.execute_write("create table Inventory (id integer primary key)")
+        await ds.invoke_startup()
+        assert not await ds.allowed(
+            action="view-table", resource=TableResource("data", "Inventory")
+        )
+        assert not (
+            await ds.allowed_resources(
+                "view-table", parent="data", include_is_private=True
+            )
+        ).resources
+        explanation = await explain_permission_for_resource(
+            datasette=ds,
+            actor=None,
+            action="view-table",
+            parent="data",
+            child="Inventory",
+        )
+        assert not explanation["allowed"]
+        assert any(
+            rule["effect"] == "allow" and not rule["decisive"]
+            for rule in explanation["matched_rules"]
+        )
+        assert any(
+            rule["effect"] == "deny" and rule["decisive"]
+            for rule in explanation["matched_rules"]
+        )
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("config_style", ["allow", "permissions"])
+@pytest.mark.parametrize("allowed", [True, False])
+async def test_case_variant_token_restrictions(config_style, allowed):
+    table_config = (
+        {"allow": allowed}
+        if config_style == "allow"
+        else {"permissions": {"view-table": allowed}}
+    )
+    ds = Datasette(
+        config={"databases": {"data": {"tables": {"Inventory": table_config}}}}
+    )
+    db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data")
+    actor = {"id": "reader", "_r": {"r": {"data": {"inventory": ["vt"]}}}}
+    try:
+        await db.execute_write("create table Inventory (id integer primary key)")
+        await ds.invoke_startup()
+        assert restrictions_allow_action(
+            ds, actor["_r"], "view-table", ("data", "INVENTORY")
+        )
+        assert not restrictions_allow_action(
+            ds, actor["_r"], "view-table", ("Data", "Inventory")
+        )
+        assert (
+            await ds.allowed(
+                action="view-table",
+                resource=TableResource("data", "INVENTORY"),
+                actor=actor,
+            )
+            is allowed
+        )
+        page = await ds.allowed_resources("view-table", actor, parent="data")
+        assert [(r.parent, r.child) for r in page.resources] == (
+            [("data", "Inventory")] if allowed else []
+        )
+        explanation = await explain_permission_for_resource(
+            datasette=ds,
+            actor=actor,
+            action="view-table",
+            parent="data",
+            child="Inventory",
+        )
+        assert explanation["restriction_allowed"]
+        assert explanation["allowed"] is allowed
+    finally:
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_plugin_restriction_intersection_and_dependencies():
+    class Plugin:
+        @hookimpl
+        def register_actions(self, datasette):
+            return [
+                Action(
+                    name="inspect-inventory",
+                    description="Inspect inventory",
+                    resource_class=TableResource,
+                    also_requires="view-table",
+                )
+            ]
+
+        @hookimpl
+        def permission_resources_sql(self, action):
+            if action not in ("view-table", "inspect-inventory"):
+                return None
+            return [
+                PermissionSQL(
+                    sql="SELECT 'data' AS parent, 'INVENTORY' AS child, 1 AS allow, 'inventory grant' AS reason",
+                    restriction_sql="SELECT 'data' AS parent, 'inventory' AS child",
+                ),
+                PermissionSQL(
+                    restriction_sql="SELECT 'data' AS parent, 'InVeNtOrY' AS child"
+                ),
+            ]
+
+    ds = Datasette(default_deny=True)
+    ds.pm.register(Plugin(), name="identity-test")
+    db = ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data")
+    try:
+        await db.execute_write("create table Inventory (id integer primary key)")
+        await db.execute_write("create table Other (id integer primary key)")
+        await ds.invoke_startup()
+        for action in ("view-table", "inspect-inventory"):
+            assert await ds.allowed(
+                action=action, resource=TableResource("data", "Inventory")
+            )
+            assert not await ds.allowed(
+                action=action, resource=TableResource("data", "Other")
+            )
+            resources = (
+                await ds.allowed_resources(
+                    action, parent="data", include_is_private=True
+                )
+            ).resources
+            assert [r.child for r in resources] == ["Inventory"]
+            explanation = await explain_permission_for_resource(
+                datasette=ds,
+                actor=None,
+                action=action,
+                parent="data",
+                child="Inventory",
+            )
+            assert explanation["allowed"]
+            assert all(item["allowed"] for item in explanation["restrictions"])
+    finally:
+        ds.pm.unregister(name="identity-test")
+        ds.close()
+
+
+@pytest.mark.asyncio
+async def test_shared_plugin_rule_keeps_query_identity_and_original_sql():
+    shared = PermissionSQL(
+        sql="SELECT 'data' AS parent, 'Inventory' AS child, 0 AS allow, 'shared deny' AS reason"
+    )
+    original_sql = shared.sql
+
+    class Plugin:
+        @hookimpl
+        def permission_resources_sql(self, action):
+            if action in ("view-table", "view-query"):
+                return shared
+
+    ds = Datasette(
+        config={
+            "databases": {
+                "data": {
+                    "queries": {
+                        "Inventory": "select 1",
+                        "inventory": "select 1",
+                    }
+                }
+            }
+        }
+    )
+    ds.add_memory_database("identity_" + uuid.uuid4().hex, name="data")
+    ds.pm.register(Plugin(), name="identity-test")
+    try:
+        await ds.invoke_startup()
+        for _ in range(2):
+            await gather_permission_sql_from_hooks(
+                datasette=ds, actor=None, action="view-table"
+            )
+        assert shared.sql == original_sql
+        assert not await ds.allowed(
+            action="view-table", resource=TableResource("data", "inventory")
+        )
+        assert await ds.allowed(
+            action="view-query", resource=QueryResource("data", "inventory")
+        )
+        assert not await ds.allowed(
+            action="view-query", resource=QueryResource("data", "Inventory")
+        )
+        assert await ds.allowed(
+            action="view-table", resource=TableResource("Data", "Inventory")
+        )
+        assert restrictions_allow_action(
+            ds,
+            {"r": {"data": {"Inventory": ["vq"]}}},
+            "view-query",
+            ("data", "Inventory"),
+        )
+        assert not restrictions_allow_action(
+            ds,
+            {"r": {"data": {"Inventory": ["vq"]}}},
+            "view-query",
+            ("data", "inventory"),
+        )
+    finally:
+        ds.pm.unregister(name="identity-test")
+        ds.close()

tests/test_utils.py

@@ -16,6 +16,7 @@ from datasette.app import Datasette
 from datasette.utils.asgi import Request
 from datasette.utils.sqlite import (
     sqlite3,
+    sqlite_derived_table_dependencies,
     sqlite_hidden_table_names,
     sqlite_table_type,
     supports_returning,
@@ -369,6 +370,46 @@ def test_sqlite_hidden_table_names_hides_multiline_content_fts_table():
         conn.close()
 
 
+def test_sqlite_derived_table_dependencies():
+    conn = utils.sqlite3.connect(":memory:")
+    try:
+        conn.executescript("""
+            create table docs(id integer primary key, body text);
+            create virtual table external_fts5 using fts5(
+                body, content='docs', content_rowid='id'
+            );
+            create virtual table internal_fts5 using fts5(body);
+            create virtual table contentless_fts5 using fts5(body, content='');
+            create virtual table external_fts4 using fts4(body, content="docs");
+            create virtual table internal_fts4 using fts4(body);
+            create virtual table contentless_fts4 using fts4(body, content="");
+            create table [docs, archive](body text);
+            create virtual table commented_fts5 using fts5(
+                body, tokenize='porter unicode61',
+                /* Comments and commas in quoted values must not confuse parsing. */
+                content='docs, archive'
+            );
+            create virtual table boxes using rtree(id, minx, maxx, miny, maxy);
+        """)
+
+        dependencies = sqlite_derived_table_dependencies(conn)
+
+        assert dependencies["external_fts5"] == "docs"
+        assert dependencies["external_fts4"] == "docs"
+        assert dependencies["commented_fts5"] == "docs, archive"
+        assert "contentless_fts5" not in dependencies
+        assert "contentless_fts4" not in dependencies
+        assert dependencies["internal_fts5_content"] == "internal_fts5"
+        assert dependencies["internal_fts4_content"] == "internal_fts4"
+        assert dependencies["external_fts5_data"] == "external_fts5"
+        assert dependencies["external_fts4_segments"] == "external_fts4"
+        assert dependencies["boxes_node"] == "boxes"
+        assert dependencies["boxes_parent"] == "boxes"
+        assert dependencies["boxes_rowid"] == "boxes"
+    finally:
+        conn.close()
+
+
 @pytest.mark.parametrize(
     "url,expected",
     [

tests/test_utils_sql_analysis.py

@@ -439,7 +439,7 @@ def test_analyze_attached_database_tables(conn):
     }
 
 
-def test_analyze_clears_authorizer_on_error():
+def test_analyze_disables_authorizer_on_error():
     class FakeConnection:
         def __init__(self):
             self.authorizers = []
@@ -455,4 +455,5 @@ def test_analyze_clears_authorizer_on_error():
     with pytest.raises(sqlite3.OperationalError):
         analyze_sql_tables(conn, "bad SQL")
 
-    assert conn.authorizers[-1] is None
+    final_authorizer = conn.authorizers[-1]
+    assert final_authorizer is None or final_authorizer() == sqlite3.SQLITE_OK