fix(ci): remove language_version: python3.10 from mypy pre-commit hook

CI runners have Python 3.11, not 3.10 — specifying language_version: python3.10
causes pre-commit to fail with CalledProcessError when building the mypy
virtualenv. Master doesn't set this; omitting it lets pre-commit use whatever
Python 3.x is available, which handles both local (3.10) and CI (3.11).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

.pre-commit-config.yaml

@@ -139,7 +139,7 @@ repos:
             BASE=$(git merge-base origin/"$TARGET_BRANCH" HEAD 2>/dev/null) || BASE="HEAD"
             files=$(git diff --name-only --diff-filter=ACM "$BASE"..HEAD 2>/dev/null | grep '^superset/.*\.py$' || true)
             if [ -n "$files" ]; then
-              pylint --rcfile=.pylintrc --load-plugins=superset.extensions.pylint --reports=no $files
+              PYTHONPATH="${PYTHONPATH:+$PYTHONPATH:}$(pwd)" pylint --rcfile=.pylintrc --load-plugins=superset.extensions.pylint --reports=no $files
             else
               echo "No Python files to lint."
             fi

UPDATING.md

@@ -32,6 +32,36 @@ The Deck.gl MapBox chart's **Opacity**, **Default longitude**, **Default latitud
 
 **To restore fit-to-data behavior:** Open the chart in Explore, clear the **Default longitude**, **Default latitude**, and **Zoom** fields in the Viewport section, and re-save the chart.
 
+### Tier-3 Extension Storage: `ExtensionStorage` model + `ExtensionStorageDAO`
+
+A generic `extension_storage` table now provides extensions with scoped,
+encrypted persistent key-value storage without requiring extensions to
+contribute model code to core Superset.
+
+Three storage scopes are supported:
+
+| Scope    | Description |
+|----------|-------------|
+| **Global** | Shared across all users of an extension |
+| **User**   | Isolated per authenticated user |
+| **Resource** | Tied to a specific Superset resource (dashboard, chart, …) by type + UUID |
+
+REST API (all endpoints under `/api/v1/extensions/{publisher}/{name}/`):
+
+```
+GET/PUT/DELETE .../storage/persistent/global/{key}
+GET            .../storage/persistent/global/         (paginated)
+GET/PUT/DELETE .../storage/persistent/user/{key}
+GET            .../storage/persistent/user/           (paginated, ?category=…)
+GET/PUT/DELETE .../storage/persistent/resources/{type}/{uuid}/{key}
+GET            .../storage/persistent/resources/{type}/{uuid}/  (paginated)
+```
+
+Values may be stored encrypted at rest by including `"is_encrypted": true` in
+the PUT body. The DAO uses a Fernet key derived from the app `SECRET_KEY`.
+Key rotation is supported via `EXTENSION_STORAGE_ENCRYPTION_KEYS` in config
+and the `superset rotate-extension-storage-keys` CLI command.
+
 ### ClickHouse minimum driver version bump
 
 The minimum required version of `clickhouse-connect` has been raised to `>=0.13.0`. If you are using the ClickHouse connector, please upgrade your `clickhouse-connect` package. The `_mutate_label` workaround that appended hash suffixes to column aliases has also been removed, as it is no longer needed with modern versions of the driver.

docs/developer_docs/extensions/extension-points/persistent-storage.md

@@ -0,0 +1,235 @@
+---
+title: Persistent Storage
+sidebar_position: 3
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Persistent Storage (Tier 3)
+
+Extensions can store arbitrary key-value data in Superset's database without
+contributing model code to the core codebase. The storage layer supports three
+**scopes**, optional **category** filtering for cheap listing, optional
+**at-rest encryption**, and **key rotation** with zero downtime.
+
+## Storage Scopes
+
+| Scope | When to use | `user_fk` | `resource_type` |
+|---|---|---|---|
+| **Global** | Configuration shared across all users of the extension | `null` | `null` |
+| **User** | Per-user preferences or saved artefacts | set | `null` |
+| **Resource** | State tied to a specific Superset resource (dashboard, chart, …) | any | set |
+
+## REST API
+
+All endpoints live under:
+
+```
+/api/v1/extensions/{publisher}/{name}/storage/persistent/
+```
+
+### Global scope
+
+| Method | Path | Description |
+|--------|------|-------------|
+| `GET` | `.../global/` | List entries (`?page=0&page_size=25&category=…`) |
+| `GET` | `.../global/{key}` | Read a value (response body is the raw bytes) |
+| `PUT` | `.../global/{key}` | Create or update |
+| `DELETE` | `.../global/{key}` | Delete |
+
+### User scope
+
+| Method | Path | Description |
+|--------|------|-------------|
+| `GET` | `.../user/` | List the authenticated user's entries |
+| `GET` | `.../user/{key}` | Read a value |
+| `PUT` | `.../user/{key}` | Create or update |
+| `DELETE` | `.../user/{key}` | Delete |
+
+### Resource scope
+
+| Method | Path | Description |
+|--------|------|-------------|
+| `GET` | `.../resources/{type}/{uuid}/` | List entries for the resource |
+| `GET` | `.../resources/{type}/{uuid}/{key}` | Read a value |
+| `PUT` | `.../resources/{type}/{uuid}/{key}` | Create or update |
+| `DELETE` | `.../resources/{type}/{uuid}/{key}` | Delete |
+
+## PUT request body
+
+All write endpoints accept a JSON body:
+
+```json
+{
+  "value": "<string or base64 bytes>",
+  "value_type": "application/json",
+  "category": "my-category",
+  "description": "Human-readable label for cheap listing",
+  "is_encrypted": false
+}
+```
+
+| Field | Required | Default | Notes |
+|---|---|---|---|
+| `value` | yes | `""` | The payload to store |
+| `value_type` | no | `application/json` | MIME type returned on GET |
+| `category` | no | `null` | Used to filter list results |
+| `description` | no | `null` | Returned in list results without fetching `value` |
+| `is_encrypted` | no | `false` | Encrypts `value` at rest with Fernet |
+
+## List response
+
+```json
+{
+  "count": 42,
+  "result": [
+    {
+      "key": "my-key",
+      "uuid": "...",
+      "value_type": "application/json",
+      "category": "my-category",
+      "description": "...",
+      "is_encrypted": false
+    }
+  ]
+}
+```
+
+The `value` blob is **not** included in list responses. Fetch individual keys
+to retrieve values.
+
+## Frontend usage
+
+From a TypeScript extension frontend, use standard `fetch` with the Superset
+CSRF token:
+
+```typescript
+const BASE = '/api/v1/extensions/acme/my-ext/storage/persistent/user';
+
+// Save
+await fetch(`${BASE}/my-key`, {
+  method: 'PUT',
+  headers: {
+    'Content-Type': 'application/json',
+    'X-CSRFToken': await authentication.getCSRFToken(),
+  },
+  body: JSON.stringify({
+    value: JSON.stringify({ foo: 'bar' }),
+    value_type: 'application/json',
+    category: 'preferences',
+    description: 'My display name',  // shown in list without fetching value
+  }),
+});
+
+// List
+const { result } = await fetch(`${BASE}/?category=preferences`).then(r => r.json());
+
+// Read
+const raw = await fetch(`${BASE}/my-key`).then(r => r.text());
+const data = JSON.parse(raw);
+
+// Delete
+await fetch(`${BASE}/my-key`, { method: 'DELETE', headers: { 'X-CSRFToken': ... } });
+```
+
+## Backend usage (Python)
+
+Within extension backend code, use `ExtensionStorageDAO` directly:
+
+```python
+from superset_core.common.daos import ExtensionStorageDAO
+
+EXT_ID = "acme.my-ext"
+
+# Write
+ExtensionStorageDAO.set(EXT_ID, "config", b'{"theme":"dark"}', category="global-config")
+
+# Read
+value: bytes | None = ExtensionStorageDAO.get_value(EXT_ID, "config")
+
+# User-scoped
+ExtensionStorageDAO.set(EXT_ID, "prefs", b'{"lang":"fr"}', user_fk=user_id)
+value = ExtensionStorageDAO.get_value(EXT_ID, "prefs", user_fk=user_id)
+
+# Resource-scoped
+ExtensionStorageDAO.set(
+    EXT_ID, "state", payload,
+    resource_type="dashboard", resource_uuid=str(dashboard.uuid),
+)
+```
+
+:::note
+`ExtensionStorageDAO` methods call `db.session.flush()` but do **not** commit.
+The caller (or the `@transaction()` decorator on the API endpoint) owns the
+transaction.
+:::
+
+## Encryption at rest
+
+Set `"is_encrypted": true` in the PUT body (frontend) or `is_encrypted=True`
+in `ExtensionStorageDAO.set()` (backend). Values are encrypted with
+[Fernet](https://cryptography.io/en/latest/fernet/) using a key derived from
+`SECRET_KEY` (or the first entry of `EXTENSION_STORAGE_ENCRYPTION_KEYS` if
+configured). Decryption is transparent on read.
+
+### Key rotation
+
+To rotate encryption keys without downtime:
+
+1. **Prepend** the new key to `EXTENSION_STORAGE_ENCRYPTION_KEYS` in your
+   Superset config and **restart**. Existing encrypted rows are still readable
+   (old key is tried as a fallback); new writes use the new key.
+
+   ```python
+   # superset_config.py
+   EXTENSION_STORAGE_ENCRYPTION_KEYS = [
+       "new-strong-secret",   # used for new encryptions
+       "old-secret",          # still tried for decryption
+   ]
+   ```
+
+2. **Re-encrypt** all existing rows with the new key:
+
+   ```bash
+   superset rotate-extension-storage-keys
+   ```
+
+3. **Remove** the old key from the list and restart again.
+
+## Data model
+
+The `extension_storage` table schema:
+
+| Column | Type | Notes |
+|---|---|---|
+| `id` | integer PK | |
+| `uuid` | UUID | Unique, auto-generated |
+| `extension_id` | varchar(255) | `{publisher}.{name}` |
+| `user_fk` | integer FK | `null` = global or resource scope |
+| `resource_type` | varchar(64) | e.g. `dashboard`, `chart` |
+| `resource_uuid` | varchar(36) | UUID of the linked resource |
+| `key` | varchar(255) | Identifier within a scope |
+| `category` | varchar(64) | Optional grouping tag |
+| `description` | text | Summary for list endpoints |
+| `value` | binary (up to 16 MB) | The stored payload |
+| `value_type` | varchar(255) | MIME type, default `application/json` |
+| `is_encrypted` | boolean | Whether `value` is Fernet-encrypted |
+| `created_on`, `changed_on` | datetime | Audit fields |
+| `created_by_fk`, `changed_by_fk` | integer FK | Audit fields |

pyproject.toml

@@ -243,6 +243,7 @@ multi_line_output = 3
 order_by_type = false
 
 [tool.mypy]
+python_version = "3.10"
 check_untyped_defs = true
 disallow_any_generics = true
 disallow_untyped_calls = true

superset/cli/update.py

@@ -101,6 +101,54 @@ def update_api_docs() -> None:
         click.secho("API version not found", err=True)
 
 
+@click.command()
+@with_appcontext
+@transaction()
+def rotate_extension_storage_keys() -> None:
+    """Re-encrypt all encrypted extension storage rows with the current key.
+
+    Run this after prepending a new key to EXTENSION_STORAGE_ENCRYPTION_KEYS
+    in your Superset config.  MultiFernet.rotate() decrypts each token with
+    whichever old key succeeds, then re-encrypts it with the first (new) key.
+    Once all rows are rotated you can remove old keys from the config list.
+    """
+    # pylint: disable=import-outside-toplevel
+    from superset import db
+    from superset.extension_storage.daos import _fernet
+    from superset.extension_storage.models import ExtensionStorage
+
+    fernet = _fernet()
+    entries = (
+        db.session.query(ExtensionStorage)
+        .filter(ExtensionStorage.is_encrypted.is_(True))
+        .all()
+    )
+    if not entries:
+        click.secho("No encrypted entries found.", fg="yellow")
+        return
+    failed = 0
+    for entry in entries:
+        try:
+            entry.value = fernet.rotate(entry.value)
+        except Exception as exc:  # pylint: disable=broad-except
+            click.secho(
+                f"Failed to rotate key={entry.key} ext={entry.extension_id}: {exc}",
+                fg="red",
+                err=True,
+            )
+            failed += 1
+    rotated = len(entries) - failed
+    click.secho(
+        f"Rotated {rotated}/{len(entries)} entries"
+        + (
+            f" ({failed} failed — check EXTENSION_STORAGE_ENCRYPTION_KEYS)"
+            if failed
+            else ""
+        ),
+        fg="green" if not failed else "yellow",
+    )
+
+
 @click.command()
 @with_appcontext
 @click.option(

superset/config.py

@@ -233,6 +233,14 @@ HASH_ALGORITHM_FALLBACKS: list[Literal["md5", "sha256"]] = ["md5"]
 # a sufficiently random sequence, ex: openssl rand -base64 42"
 SECRET_KEY = os.environ.get("SUPERSET_SECRET_KEY") or CHANGE_ME_SECRET_KEY
 
+# Keys used to encrypt extension storage entries (Tier 3 persistent storage).
+# The first key in the list is used for new encryptions; all keys are tried
+# on decryption.  To rotate: prepend the new key, restart, then run
+# ``superset rotate-extension-storage-keys``.  Once all rows are re-encrypted
+# the old key(s) can be removed from the list.
+# Falls back to SECRET_KEY when not set.
+EXTENSION_STORAGE_ENCRYPTION_KEYS: list[str] = []
+
 # The SQLAlchemy connection string.
 SQLALCHEMY_DATABASE_URI = (
     f"""sqlite:///{os.path.join(DATA_DIR, "superset.db")}?check_same_thread=false"""

superset/core/api/core_api_injection.py

@@ -58,6 +58,9 @@ def inject_dao_implementations() -> None:
     from superset.daos.tag import TagDAO as HostTagDAO
     from superset.daos.tasks import TaskDAO as HostTaskDAO
     from superset.daos.user import UserDAO as HostUserDAO
+    from superset.extension_storage.daos import (
+        ExtensionStorageDAO as HostExtensionStorageDAO,
+    )
 
     # Replace abstract classes in common.daos with concrete implementations
     core_common_dao_module.DatasetDAO = HostDatasetDAO  # type: ignore[assignment,misc]
@@ -67,6 +70,7 @@ def inject_dao_implementations() -> None:
     core_common_dao_module.UserDAO = HostUserDAO  # type: ignore[assignment,misc]
     core_common_dao_module.TagDAO = HostTagDAO  # type: ignore[assignment,misc]
     core_common_dao_module.KeyValueDAO = HostKeyValueDAO  # type: ignore[assignment,misc]
+    core_common_dao_module.ExtensionStorageDAO = HostExtensionStorageDAO  # type: ignore[assignment,misc]
 
     # Replace abstract classes in queries.daos
     core_queries_dao_module.QueryDAO = HostQueryDAO  # type: ignore[assignment,misc]
@@ -89,6 +93,9 @@ def inject_model_implementations() -> None:
     from flask_appbuilder.security.sqla.models import User as HostUser
 
     from superset.connectors.sqla.models import SqlaTable as HostDataset
+    from superset.extension_storage.models import (
+        ExtensionStorage as HostExtensionStorage,
+    )
     from superset.key_value.models import KeyValueEntry as HostKeyValue
     from superset.models.core import Database as HostDatabase
     from superset.models.dashboard import Dashboard as HostDashboard
@@ -105,6 +112,7 @@ def inject_model_implementations() -> None:
     core_common_models_module.User = HostUser  # type: ignore[misc]
     core_common_models_module.Tag = HostTag  # type: ignore[misc]
     core_common_models_module.KeyValue = HostKeyValue  # type: ignore[misc]
+    core_common_models_module.ExtensionStorage = HostExtensionStorage  # type: ignore[misc]
 
     # In-place replacement in queries.models
     core_queries_models_module.Query = HostQuery  # type: ignore[misc]

superset/extension_storage/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

superset/extension_storage/api.py

@@ -0,0 +1,719 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""REST API for extension Tier-3 persistent storage.
+
+URL structure (all under /api/v1/extensions/<publisher>/<name>/storage/persistent/):
+
+  GET             /global/
+      Paginated list of global keys (?page=0&page_size=25&category=…).
+
+  GET/PUT/DELETE  /global/<key>
+      Global key-value: shared across all users of this extension.
+
+  GET             /user/
+      Paginated list of user-scoped keys (?page=0&page_size=25&category=…).
+
+  GET/PUT/DELETE  /user/<key>
+      User-scoped key-value: isolated per authenticated user.
+
+  GET             /resources/<resource_type>/<resource_uuid>/
+      Paginated list of resource-linked keys (?page=0&page_size=25&category=…).
+
+  GET/PUT/DELETE  /resources/<resource_type>/<resource_uuid>/<key>
+      Resource-linked key-value: tied to a specific Superset resource.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from flask import request, Response
+from flask_appbuilder.api import expose, protect, safe
+
+from superset.extension_storage.daos import ExtensionStorageDAO
+from superset.extension_storage.models import ExtensionStorage
+from superset.superset_typing import FlaskResponse
+from superset.utils.core import get_user_id
+from superset.utils.decorators import transaction
+from superset.views.base_api import BaseSupersetApi
+
+logger = logging.getLogger(__name__)
+
+_MIME_DEFAULT = "application/json"
+_PAGE_SIZE_DEFAULT = 25
+_PAGE_SIZE_MAX = 100
+
+
+def _parse_pagination() -> tuple[int, int]:
+    """Return (offset, page_size) from ?page=&page_size= query params."""
+    try:
+        page = max(0, int(request.args.get("page", 0)))
+    except (TypeError, ValueError):
+        page = 0
+    try:
+        page_size = min(
+            _PAGE_SIZE_MAX,
+            max(1, int(request.args.get("page_size", _PAGE_SIZE_DEFAULT))),
+        )
+    except (TypeError, ValueError):
+        page_size = _PAGE_SIZE_DEFAULT
+    return page * page_size, page_size
+
+
+def _extension_id(publisher: str, name: str) -> str:
+    return f"{publisher}.{name}"
+
+
+def _entry_to_dict(entry: "ExtensionStorage") -> dict[str, object]:
+    return {
+        "key": entry.key,
+        "value_type": entry.value_type,
+        "category": entry.category,
+        "description": entry.description,
+        "is_encrypted": entry.is_encrypted,
+        "uuid": str(entry.uuid),
+    }
+
+
+class ExtensionStorageRestApi(BaseSupersetApi):
+    """Generic persistent storage for extensions (Tier 3)."""
+
+    allow_browser_login = True
+    route_base = "/api/v1/extensions"
+    resource_name = "extensions"
+    openapi_spec_tag = "Extension Storage"
+
+    # ── Global scope ──────────────────────────────────────────────────────────
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/global/",
+        methods=("GET",),
+    )
+    @protect()
+    @safe
+    def list_global(self, publisher: str, name: str) -> FlaskResponse:
+        """List global (non-user-scoped) storage entries with pagination.
+        ---
+        get:
+          summary: List global persistent storage entries
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: category
+              in: query
+              required: false
+              schema: {type: string}
+            - name: page
+              in: query
+              required: false
+              schema: {type: integer, default: 0}
+            - name: page_size
+              in: query
+              required: false
+              schema: {type: integer, default: 25}
+          responses:
+            200:
+              description: Success
+        """
+        ext_id = _extension_id(publisher, name)
+        category = request.args.get("category")
+        offset, page_size = _parse_pagination()
+        entries = ExtensionStorageDAO.list_global(ext_id, category=category)
+        total = len(entries)
+        page_entries = entries[offset : offset + page_size]
+        return self.response(
+            200,
+            result=[_entry_to_dict(e) for e in page_entries],
+            count=total,
+        )
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/global/<key>",
+        methods=("GET",),
+    )
+    @protect()
+    @safe
+    def get_global(self, publisher: str, name: str, key: str) -> FlaskResponse:
+        """Get a global (non-user-scoped) persistent value.
+        ---
+        get:
+          summary: Get global persistent value
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: key
+              in: path
+              required: true
+              schema: {type: string}
+          responses:
+            200:
+              description: Success
+            404:
+              description: Not found
+        """
+        ext_id = _extension_id(publisher, name)
+        entry = ExtensionStorageDAO.get(ext_id, key)
+        if entry is None:
+            return self.response(404, message="Not found")
+        value = ExtensionStorageDAO.get_value(ext_id, key) or b""
+        return Response(value, status=200, mimetype=entry.value_type)
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/global/<key>",
+        methods=("PUT",),
+    )
+    @protect()
+    @safe
+    @transaction()
+    def set_global(self, publisher: str, name: str, key: str) -> FlaskResponse:
+        """Create or update a global persistent value.
+        ---
+        put:
+          summary: Set global persistent value
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: key
+              in: path
+              required: true
+              schema: {type: string}
+          requestBody:
+            content:
+              application/json:
+                schema:
+                  type: object
+                  properties:
+                    value: {type: string}
+                    value_type: {type: string}
+                    category: {type: string}
+                    description: {type: string}
+                    is_encrypted: {type: boolean}
+          responses:
+            200:
+              description: Success
+        """
+        ext_id = _extension_id(publisher, name)
+        body = request.get_json(force=True) or {}
+        raw_value: str | bytes = body.get("value", "")
+        value_bytes = raw_value.encode() if isinstance(raw_value, str) else raw_value
+        entry = ExtensionStorageDAO.set(
+            extension_id=ext_id,
+            key=key,
+            value=value_bytes,
+            value_type=body.get("value_type", _MIME_DEFAULT),
+            category=body.get("category"),
+            description=body.get("description"),
+            is_encrypted=bool(body.get("is_encrypted", False)),
+        )
+        return self.response(200, result=_entry_to_dict(entry))
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/global/<key>",
+        methods=("DELETE",),
+    )
+    @protect()
+    @safe
+    @transaction()
+    def delete_global(self, publisher: str, name: str, key: str) -> FlaskResponse:
+        """Delete a global persistent value.
+        ---
+        delete:
+          summary: Delete global persistent value
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: key
+              in: path
+              required: true
+              schema: {type: string}
+          responses:
+            200:
+              description: Deleted
+            404:
+              description: Not found
+        """
+        ext_id = _extension_id(publisher, name)
+        deleted = ExtensionStorageDAO.delete(ext_id, key)
+        if not deleted:
+            return self.response(404, message="Not found")
+        return self.response(200, message="Deleted")
+
+    # ── User scope ────────────────────────────────────────────────────────────
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/user/",
+        methods=("GET",),
+    )
+    @protect()
+    @safe
+    def list_user(self, publisher: str, name: str) -> FlaskResponse:
+        """List user-scoped storage entries for the current user with pagination.
+        ---
+        get:
+          summary: List user-scoped storage entries
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: category
+              in: query
+              required: false
+              schema: {type: string}
+            - name: page
+              in: query
+              required: false
+              schema: {type: integer, default: 0}
+            - name: page_size
+              in: query
+              required: false
+              schema: {type: integer, default: 25}
+          responses:
+            200:
+              description: Success
+        """
+        user_id = get_user_id()
+        if not user_id:
+            return self.response(401, message="Authentication required")
+        ext_id = _extension_id(publisher, name)
+        category = request.args.get("category")
+        offset, page_size = _parse_pagination()
+        entries = ExtensionStorageDAO.list_user(ext_id, user_id, category=category)
+        total = len(entries)
+        page_entries = entries[offset : offset + page_size]
+        return self.response(
+            200, result=[_entry_to_dict(e) for e in page_entries], count=total
+        )
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/user/<key>",
+        methods=("GET",),
+    )
+    @protect()
+    @safe
+    def get_user(self, publisher: str, name: str, key: str) -> FlaskResponse:
+        """Get a user-scoped persistent value.
+        ---
+        get:
+          summary: Get user-scoped persistent value
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: key
+              in: path
+              required: true
+              schema: {type: string}
+          responses:
+            200:
+              description: Success
+            404:
+              description: Not found
+        """
+        user_id = get_user_id()
+        if not user_id:
+            return self.response(401, message="Authentication required")
+        ext_id = _extension_id(publisher, name)
+        entry = ExtensionStorageDAO.get(ext_id, key, user_fk=user_id)
+        if entry is None:
+            return self.response(404, message="Not found")
+        value = ExtensionStorageDAO.get_value(ext_id, key, user_fk=user_id) or b""
+        return Response(value, status=200, mimetype=entry.value_type)
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/user/<key>",
+        methods=("PUT",),
+    )
+    @protect()
+    @safe
+    @transaction()
+    def set_user(self, publisher: str, name: str, key: str) -> FlaskResponse:
+        """Create or update a user-scoped persistent value.
+        ---
+        put:
+          summary: Set user-scoped persistent value
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: key
+              in: path
+              required: true
+              schema: {type: string}
+          requestBody:
+            content:
+              application/json:
+                schema:
+                  type: object
+                  properties:
+                    value: {type: string}
+                    value_type: {type: string}
+                    category: {type: string}
+                    description: {type: string}
+                    is_encrypted: {type: boolean}
+          responses:
+            200:
+              description: Success
+            401:
+              description: Authentication required
+        """
+        user_id = get_user_id()
+        if not user_id:
+            return self.response(401, message="Authentication required")
+        ext_id = _extension_id(publisher, name)
+        body = request.get_json(force=True) or {}
+        raw_value: str | bytes = body.get("value", "")
+        value_bytes = raw_value.encode() if isinstance(raw_value, str) else raw_value
+        entry = ExtensionStorageDAO.set(
+            extension_id=ext_id,
+            key=key,
+            value=value_bytes,
+            value_type=body.get("value_type", _MIME_DEFAULT),
+            user_fk=user_id,
+            category=body.get("category"),
+            description=body.get("description"),
+            is_encrypted=bool(body.get("is_encrypted", False)),
+        )
+        return self.response(200, result=_entry_to_dict(entry))
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/user/<key>",
+        methods=("DELETE",),
+    )
+    @protect()
+    @safe
+    @transaction()
+    def delete_user(self, publisher: str, name: str, key: str) -> FlaskResponse:
+        """Delete a user-scoped persistent value.
+        ---
+        delete:
+          summary: Delete user-scoped persistent value
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: key
+              in: path
+              required: true
+              schema: {type: string}
+          responses:
+            200:
+              description: Deleted
+            404:
+              description: Not found
+        """
+        user_id = get_user_id()
+        if not user_id:
+            return self.response(401, message="Authentication required")
+        ext_id = _extension_id(publisher, name)
+        deleted = ExtensionStorageDAO.delete(ext_id, key, user_fk=user_id)
+        if not deleted:
+            return self.response(404, message="Not found")
+        return self.response(200, message="Deleted")
+
+    # ── Resource scope ────────────────────────────────────────────────────────
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/resources/<resource_type>/<resource_uuid>/",
+        methods=("GET",),
+    )
+    @protect()
+    @safe
+    def list_resource(
+        self,
+        publisher: str,
+        name: str,
+        resource_type: str,
+        resource_uuid: str,
+    ) -> FlaskResponse:
+        """List storage entries linked to a resource with pagination.
+        ---
+        get:
+          summary: List resource-linked storage entries
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: resource_type
+              in: path
+              required: true
+              schema: {type: string}
+            - name: resource_uuid
+              in: path
+              required: true
+              schema: {type: string}
+            - name: category
+              in: query
+              required: false
+              schema: {type: string}
+            - name: page
+              in: query
+              required: false
+              schema: {type: integer, default: 0}
+            - name: page_size
+              in: query
+              required: false
+              schema: {type: integer, default: 25}
+          responses:
+            200:
+              description: Success
+        """
+        ext_id = _extension_id(publisher, name)
+        category = request.args.get("category")
+        offset, page_size = _parse_pagination()
+        entries = ExtensionStorageDAO.list_resource(
+            ext_id, resource_type, resource_uuid, category=category
+        )
+        total = len(entries)
+        page_entries = entries[offset : offset + page_size]
+        return self.response(
+            200, result=[_entry_to_dict(e) for e in page_entries], count=total
+        )
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/resources/<resource_type>/<resource_uuid>/<key>",
+        methods=("GET",),
+    )
+    @protect()
+    @safe
+    def get_resource(
+        self,
+        publisher: str,
+        name: str,
+        resource_type: str,
+        resource_uuid: str,
+        key: str,
+    ) -> FlaskResponse:
+        """Get a resource-linked persistent value.
+        ---
+        get:
+          summary: Get resource-linked persistent value
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: resource_type
+              in: path
+              required: true
+              schema: {type: string}
+            - name: resource_uuid
+              in: path
+              required: true
+              schema: {type: string}
+            - name: key
+              in: path
+              required: true
+              schema: {type: string}
+          responses:
+            200:
+              description: Success
+            404:
+              description: Not found
+        """
+        ext_id = _extension_id(publisher, name)
+        entry = ExtensionStorageDAO.get(
+            ext_id, key, resource_type=resource_type, resource_uuid=resource_uuid
+        )
+        if entry is None:
+            return self.response(404, message="Not found")
+        value = (
+            ExtensionStorageDAO.get_value(
+                ext_id, key, resource_type=resource_type, resource_uuid=resource_uuid
+            )
+            or b""
+        )
+        return Response(value, status=200, mimetype=entry.value_type)
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/resources/<resource_type>/<resource_uuid>/<key>",
+        methods=("PUT",),
+    )
+    @protect()
+    @safe
+    @transaction()
+    def set_resource(
+        self,
+        publisher: str,
+        name: str,
+        resource_type: str,
+        resource_uuid: str,
+        key: str,
+    ) -> FlaskResponse:
+        """Create or update a resource-linked persistent value.
+        ---
+        put:
+          summary: Set resource-linked persistent value
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: resource_type
+              in: path
+              required: true
+              schema: {type: string}
+            - name: resource_uuid
+              in: path
+              required: true
+              schema: {type: string}
+            - name: key
+              in: path
+              required: true
+              schema: {type: string}
+          requestBody:
+            content:
+              application/json:
+                schema:
+                  type: object
+                  properties:
+                    value: {type: string}
+                    value_type: {type: string}
+                    category: {type: string}
+                    description: {type: string}
+                    is_encrypted: {type: boolean}
+          responses:
+            200:
+              description: Success
+        """
+        ext_id = _extension_id(publisher, name)
+        body = request.get_json(force=True) or {}
+        raw_value: str | bytes = body.get("value", "")
+        value_bytes = raw_value.encode() if isinstance(raw_value, str) else raw_value
+        entry = ExtensionStorageDAO.set(
+            extension_id=ext_id,
+            key=key,
+            value=value_bytes,
+            value_type=body.get("value_type", _MIME_DEFAULT),
+            resource_type=resource_type,
+            resource_uuid=resource_uuid,
+            category=body.get("category"),
+            description=body.get("description"),
+            is_encrypted=bool(body.get("is_encrypted", False)),
+        )
+        return self.response(200, result=_entry_to_dict(entry))
+
+    @expose(
+        "/<publisher>/<name>/storage/persistent/resources/<resource_type>/<resource_uuid>/<key>",
+        methods=("DELETE",),
+    )
+    @protect()
+    @safe
+    @transaction()
+    def delete_resource(
+        self,
+        publisher: str,
+        name: str,
+        resource_type: str,
+        resource_uuid: str,
+        key: str,
+    ) -> FlaskResponse:
+        """Delete a resource-linked persistent value.
+        ---
+        delete:
+          summary: Delete resource-linked persistent value
+          parameters:
+            - name: publisher
+              in: path
+              required: true
+              schema: {type: string}
+            - name: name
+              in: path
+              required: true
+              schema: {type: string}
+            - name: resource_type
+              in: path
+              required: true
+              schema: {type: string}
+            - name: resource_uuid
+              in: path
+              required: true
+              schema: {type: string}
+            - name: key
+              in: path
+              required: true
+              schema: {type: string}
+          responses:
+            200:
+              description: Deleted
+            404:
+              description: Not found
+        """
+        ext_id = _extension_id(publisher, name)
+        deleted = ExtensionStorageDAO.delete(
+            ext_id, key, resource_type=resource_type, resource_uuid=resource_uuid
+        )
+        if not deleted:
+            return self.response(404, message="Not found")
+        return self.response(200, message="Deleted")

superset/extension_storage/daos.py

@@ -0,0 +1,268 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import logging
+
+from cryptography.fernet import Fernet, MultiFernet
+from flask import current_app
+from sqlalchemy import and_
+
+from superset import db
+from superset.extension_storage.models import ExtensionStorage
+
+logger = logging.getLogger(__name__)
+
+
+def _key_to_fernet(raw_key: str | bytes) -> Fernet:
+    """Derive a Fernet instance from an arbitrary-length secret string.
+
+    SHA-256 compresses the key to exactly 32 bytes, which are then
+    base64url-encoded to satisfy Fernet's key format requirement.
+    """
+    if isinstance(raw_key, str):
+        raw_key = raw_key.encode()
+    return Fernet(base64.urlsafe_b64encode(hashlib.sha256(raw_key).digest()))
+
+
+def _fernet() -> MultiFernet:
+    """Return a MultiFernet built from EXTENSION_STORAGE_ENCRYPTION_KEYS.
+
+    Falls back to SECRET_KEY when the config list is absent.  The first key in
+    the list is used for new encryptions; all keys are tried on decryption,
+    enabling zero-downtime rotation: add the new key at the front of
+    EXTENSION_STORAGE_ENCRYPTION_KEYS, then run ``superset rotate-extension-
+    storage-keys`` to re-encrypt every row with the new key.
+    """
+    raw_keys: list[str | bytes] = current_app.config.get(
+        "EXTENSION_STORAGE_ENCRYPTION_KEYS"
+    ) or [current_app.config["SECRET_KEY"]]
+    return MultiFernet([_key_to_fernet(k) for k in raw_keys])
+
+
+def _scope_filter(
+    extension_id: str,
+    key: str,
+    user_fk: int | None = None,
+    resource_type: str | None = None,
+    resource_uuid: str | None = None,
+) -> list[object]:
+    """Build the SQLAlchemy filter list for a scoped lookup."""
+    filters: list[object] = [
+        ExtensionStorage.extension_id == extension_id,
+        ExtensionStorage.key == key,
+    ]
+    if user_fk is None:
+        filters.append(ExtensionStorage.user_fk.is_(None))
+    else:
+        filters.append(ExtensionStorage.user_fk == user_fk)
+    if resource_type is None:
+        filters.append(ExtensionStorage.resource_type.is_(None))
+    else:
+        filters.append(ExtensionStorage.resource_type == resource_type)
+    if resource_uuid is None:
+        filters.append(ExtensionStorage.resource_uuid.is_(None))
+    else:
+        filters.append(ExtensionStorage.resource_uuid == resource_uuid)
+    return filters
+
+
+class ExtensionStorageDAO:
+    """Persistent key-value store for extensions.
+
+    Provides scoped get/set/delete and list operations covering the three
+    storage scopes defined by the Tier 3 proposal:
+
+    * Global scope      — user_fk=None, resource_type=None
+    * User scope        — user_fk=<id>, resource_type=None
+    * Resource scope    — resource_type + resource_uuid set
+    """
+
+    # ── Read ─────────────────────────────────────────────────────────────────
+
+    @staticmethod
+    def get(
+        extension_id: str,
+        key: str,
+        user_fk: int | None = None,
+        resource_type: str | None = None,
+        resource_uuid: str | None = None,
+    ) -> ExtensionStorage | None:
+        """Return a storage entry, decrypting the value if needed."""
+        entry = (
+            db.session.query(ExtensionStorage)
+            .filter(
+                and_(
+                    *_scope_filter(
+                        extension_id, key, user_fk, resource_type, resource_uuid
+                    )
+                )
+            )
+            .first()
+        )
+        return entry
+
+    @staticmethod
+    def get_value(
+        extension_id: str,
+        key: str,
+        user_fk: int | None = None,
+        resource_type: str | None = None,
+        resource_uuid: str | None = None,
+    ) -> bytes | None:
+        """Return the raw (decrypted) value bytes, or None if not found."""
+        entry = ExtensionStorageDAO.get(
+            extension_id, key, user_fk, resource_type, resource_uuid
+        )
+        if entry is None:
+            return None
+        if entry.is_encrypted:
+            return _fernet().decrypt(entry.value)
+        return entry.value
+
+    # ── Write (upsert) ────────────────────────────────────────────────────────
+
+    @staticmethod
+    def set(
+        extension_id: str,
+        key: str,
+        value: bytes,
+        value_type: str = "application/json",
+        user_fk: int | None = None,
+        resource_type: str | None = None,
+        resource_uuid: str | None = None,
+        category: str | None = None,
+        description: str | None = None,
+        is_encrypted: bool = False,
+    ) -> ExtensionStorage:
+        """Upsert a storage entry.  Encrypts value when is_encrypted=True."""
+        stored_value = _fernet().encrypt(value) if is_encrypted else value
+
+        entry = (
+            db.session.query(ExtensionStorage)
+            .filter(
+                and_(
+                    *_scope_filter(
+                        extension_id, key, user_fk, resource_type, resource_uuid
+                    )
+                )
+            )
+            .first()
+        )
+        if entry is not None:
+            entry.value = stored_value
+            entry.value_type = value_type
+            entry.category = category
+            entry.description = description
+            entry.is_encrypted = is_encrypted
+        else:
+            entry = ExtensionStorage(
+                extension_id=extension_id,
+                key=key,
+                value=stored_value,
+                value_type=value_type,
+                user_fk=user_fk,
+                resource_type=resource_type,
+                resource_uuid=resource_uuid,
+                category=category,
+                description=description,
+                is_encrypted=is_encrypted,
+            )
+            db.session.add(entry)
+        db.session.flush()
+        return entry
+
+    # ── Delete ────────────────────────────────────────────────────────────────
+
+    @staticmethod
+    def delete(
+        extension_id: str,
+        key: str,
+        user_fk: int | None = None,
+        resource_type: str | None = None,
+        resource_uuid: str | None = None,
+    ) -> bool:
+        """Delete an entry. Returns True if a row was removed."""
+        entry = (
+            db.session.query(ExtensionStorage)
+            .filter(
+                and_(
+                    *_scope_filter(
+                        extension_id, key, user_fk, resource_type, resource_uuid
+                    )
+                )
+            )
+            .first()
+        )
+        if entry is None:
+            return False
+        db.session.delete(entry)
+        db.session.flush()
+        return True
+
+    # ── List ──────────────────────────────────────────────────────────────────
+
+    @staticmethod
+    def list_global(
+        extension_id: str,
+        category: str | None = None,
+    ) -> list[ExtensionStorage]:
+        """List all global (user_fk=NULL, resource_type=NULL) entries."""
+        q = db.session.query(ExtensionStorage).filter(
+            ExtensionStorage.extension_id == extension_id,
+            ExtensionStorage.user_fk.is_(None),
+            ExtensionStorage.resource_type.is_(None),
+        )
+        if category is not None:
+            q = q.filter(ExtensionStorage.category == category)
+        return q.order_by(ExtensionStorage.key).all()
+
+    @staticmethod
+    def list_user(
+        extension_id: str,
+        user_fk: int,
+        category: str | None = None,
+    ) -> list[ExtensionStorage]:
+        """List all user-scoped entries (resource_type=NULL)."""
+        q = db.session.query(ExtensionStorage).filter(
+            ExtensionStorage.extension_id == extension_id,
+            ExtensionStorage.user_fk == user_fk,
+            ExtensionStorage.resource_type.is_(None),
+        )
+        if category is not None:
+            q = q.filter(ExtensionStorage.category == category)
+        return q.order_by(ExtensionStorage.key).all()
+
+    @staticmethod
+    def list_resource(
+        extension_id: str,
+        resource_type: str,
+        resource_uuid: str,
+        category: str | None = None,
+    ) -> list[ExtensionStorage]:
+        """List all entries linked to a specific resource."""
+        q = db.session.query(ExtensionStorage).filter(
+            ExtensionStorage.extension_id == extension_id,
+            ExtensionStorage.resource_type == resource_type,
+            ExtensionStorage.resource_uuid == resource_uuid,
+        )
+        if category is not None:
+            q = q.filter(ExtensionStorage.category == category)
+        return q.order_by(ExtensionStorage.key).all()

superset/extension_storage/models.py

@@ -0,0 +1,108 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import uuid as uuid_module
+
+from flask_appbuilder import Model
+from sqlalchemy import (
+    Boolean,
+    Column,
+    ForeignKey,
+    Index,
+    Integer,
+    LargeBinary,
+    String,
+    Text,
+)
+from sqlalchemy.orm import backref, relationship
+from sqlalchemy_utils import UUIDType
+
+from superset.models.helpers import AuditMixinNullable
+
+# 16 MB — matches the KeyValue store limit.
+EXTENSION_STORAGE_MAX_SIZE = 2**24 - 1
+
+
+class ExtensionStorage(AuditMixinNullable, Model):
+    """Generic persistent key-value storage for extensions (Tier 3).
+
+    Each row is identified by (extension_id, user_fk, resource_type,
+    resource_uuid, key):
+
+    * Global scope      — user_fk IS NULL, resource_type IS NULL
+    * User scope        — user_fk set, resource_type IS NULL
+    * Resource scope    — resource_type + resource_uuid set (user_fk optional)
+
+    The payload is stored as raw bytes (value) with a MIME-type hint
+    (value_type).  When is_encrypted is True the value has been encrypted
+    at the DAO layer using Fernet and must be decrypted before use.
+    """
+
+    __tablename__ = "extension_storage"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    uuid = Column(
+        UUIDType(binary=True),
+        default=uuid_module.uuid4,
+        unique=True,
+        nullable=False,
+    )
+
+    # Extension identity
+    extension_id = Column(String(255), nullable=False)
+
+    # Scope discriminators — all nullable; NULLs define the scope (see docstring)
+    user_fk = Column(Integer, ForeignKey("ab_user.id"), nullable=True)
+    resource_type = Column(String(64), nullable=True)
+    resource_uuid = Column(String(36), nullable=True)
+
+    # Storage key within the scope
+    key = Column(String(255), nullable=False)
+
+    # Optional metadata
+    category = Column(String(64), nullable=True)
+    description = Column(Text, nullable=True)
+
+    # Payload
+    value = Column(LargeBinary(EXTENSION_STORAGE_MAX_SIZE), nullable=False)
+    value_type = Column(String(255), nullable=False, default="application/json")
+    is_encrypted = Column(Boolean, nullable=False, default=False)
+
+    user = relationship(
+        "User",
+        backref=backref("extension_storage_entries", cascade="all, delete-orphan"),
+        foreign_keys=[user_fk],
+    )
+
+    __table_args__ = (
+        # Composite index covering all lookup dimensions
+        Index(
+            "ix_ext_storage_lookup",
+            "extension_id",
+            "user_fk",
+            "resource_type",
+            "resource_uuid",
+            "key",
+        ),
+        Index("ix_ext_storage_extension_id", "extension_id"),
+    )
+
+    def __repr__(self) -> str:
+        return (
+            f"<ExtensionStorage {self.extension_id}/"
+            f"user={self.user_fk}/res={self.resource_type}/{self.key}>"
+        )

superset/initialization/__init__.py

@@ -273,6 +273,10 @@ class SupersetAppInitializer:  # pylint: disable=too-many-public-methods
         appbuilder.add_api(SqlLabPermalinkRestApi)
         appbuilder.add_api(LogRestApi)
 
+        from superset.extension_storage.api import ExtensionStorageRestApi
+
+        appbuilder.add_api(ExtensionStorageRestApi)
+
         if feature_flag_manager.is_feature_enabled("ENABLE_EXTENSIONS"):
             from superset.extensions.api import ExtensionsRestApi
 

superset/migrations/versions/2026-03-20_12-00_e5f6a7b8c9d0_add_extension_storage_table.py

@@ -0,0 +1,84 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""add_extension_storage_table
+
+Revision ID: e5f6a7b8c9d0
+Revises: a1b2c3d4e5f6
+Create Date: 2026-03-20 12:00:00.000000
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = "e5f6a7b8c9d0"
+down_revision = "a1b2c3d4e5f6"
+
+import sqlalchemy as sa  # noqa: E402
+from alembic import op  # noqa: E402
+from sqlalchemy_utils import UUIDType  # noqa: E402
+
+
+def upgrade() -> None:
+    op.create_table(
+        "extension_storage",
+        sa.Column("id", sa.Integer(), nullable=False, autoincrement=True),
+        sa.Column("uuid", UUIDType(binary=True), nullable=False),
+        sa.Column("extension_id", sa.String(255), nullable=False),
+        sa.Column("user_fk", sa.Integer(), nullable=True),
+        sa.Column("resource_type", sa.String(64), nullable=True),
+        sa.Column("resource_uuid", sa.String(36), nullable=True),
+        sa.Column("key", sa.String(255), nullable=False),
+        sa.Column("category", sa.String(64), nullable=True),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("value", sa.LargeBinary(2**24 - 1), nullable=False),
+        sa.Column(
+            "value_type",
+            sa.String(255),
+            nullable=False,
+            server_default="application/json",
+        ),
+        sa.Column(
+            "is_encrypted",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.false(),
+        ),
+        sa.Column("created_on", sa.DateTime(), nullable=True),
+        sa.Column("changed_on", sa.DateTime(), nullable=True),
+        sa.Column("created_by_fk", sa.Integer(), nullable=True),
+        sa.Column("changed_by_fk", sa.Integer(), nullable=True),
+        sa.ForeignKeyConstraint(["user_fk"], ["ab_user.id"]),
+        sa.ForeignKeyConstraint(["created_by_fk"], ["ab_user.id"]),
+        sa.ForeignKeyConstraint(["changed_by_fk"], ["ab_user.id"]),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("uuid"),
+    )
+    op.create_index(
+        "ix_ext_storage_extension_id",
+        "extension_storage",
+        ["extension_id"],
+    )
+    op.create_index(
+        "ix_ext_storage_lookup",
+        "extension_storage",
+        ["extension_id", "user_fk", "resource_type", "resource_uuid", "key"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_ext_storage_lookup", "extension_storage")
+    op.drop_index("ix_ext_storage_extension_id", "extension_storage")
+    op.drop_table("extension_storage")

tests/unit_tests/daos/test_extension_storage_dao.py

@@ -0,0 +1,408 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Unit tests for ExtensionStorageDAO.
+
+Covers the three storage scopes (global, user, resource) and verifies that
+scope isolation, upsert semantics, and delete behaviour all work correctly.
+Encryption is tested via a mock Fernet key derived from the test SECRET_KEY.
+"""
+
+from typing import Iterator
+
+import pytest
+from sqlalchemy.orm.session import Session
+
+from superset.extension_storage.daos import ExtensionStorageDAO
+from superset.extension_storage.models import ExtensionStorage
+
+
+@pytest.fixture
+def session(session: Session) -> Iterator[Session]:
+    """Extend the base session fixture with the extension_storage table."""
+    engine = session.get_bind()
+    ExtensionStorage.metadata.create_all(engine)
+    yield session
+    session.rollback()
+
+
+EXT_A = "community.jupyter-notebook"
+EXT_B = "community.other-extension"
+USER_1 = 1
+USER_2 = 2
+RES_TYPE = "dashboard"
+RES_UUID = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+
+def _make_entry(
+    session: Session,
+    *,
+    extension_id: str = EXT_A,
+    key: str = "mykey",
+    value: bytes = b"hello",
+    value_type: str = "application/json",
+    user_fk: int | None = None,
+    resource_type: str | None = None,
+    resource_uuid: str | None = None,
+    category: str | None = None,
+) -> ExtensionStorage:
+    entry = ExtensionStorage(
+        extension_id=extension_id,
+        key=key,
+        value=value,
+        value_type=value_type,
+        user_fk=user_fk,
+        resource_type=resource_type,
+        resource_uuid=resource_uuid,
+        category=category,
+        is_encrypted=False,
+    )
+    session.add(entry)
+    session.flush()
+    return entry
+
+
+# ── Global scope ──────────────────────────────────────────────────────────────
+
+
+def test_set_and_get_global(session: Session) -> None:
+    ExtensionStorageDAO.set(EXT_A, "cfg", b'{"theme":"dark"}')
+    val = ExtensionStorageDAO.get_value(EXT_A, "cfg")
+    assert val == b'{"theme":"dark"}'
+
+
+def test_global_upsert_updates_value(session: Session) -> None:
+    ExtensionStorageDAO.set(EXT_A, "cfg", b"v1")
+    ExtensionStorageDAO.set(EXT_A, "cfg", b"v2")
+    entries = ExtensionStorageDAO.list_global(EXT_A)
+    assert len(entries) == 1
+    assert entries[0].value == b"v2"
+
+
+def test_global_delete(session: Session) -> None:
+    ExtensionStorageDAO.set(EXT_A, "cfg", b"data")
+    deleted = ExtensionStorageDAO.delete(EXT_A, "cfg")
+    assert deleted is True
+    assert ExtensionStorageDAO.get_value(EXT_A, "cfg") is None
+
+
+def test_global_delete_missing_returns_false(session: Session) -> None:
+    assert ExtensionStorageDAO.delete(EXT_A, "nonexistent") is False
+
+
+def test_global_scope_isolated_from_user_scope(session: Session) -> None:
+    """A global key and a same-named user key do not collide."""
+    ExtensionStorageDAO.set(EXT_A, "key", b"global-val")
+    ExtensionStorageDAO.set(EXT_A, "key", b"user-val", user_fk=USER_1)
+    assert ExtensionStorageDAO.get_value(EXT_A, "key") == b"global-val"
+    assert ExtensionStorageDAO.get_value(EXT_A, "key", user_fk=USER_1) == b"user-val"
+
+
+def test_list_global_filtered_by_extension(session: Session) -> None:
+    _make_entry(session, extension_id=EXT_A, key="k1")
+    _make_entry(session, extension_id=EXT_A, key="k2")
+    _make_entry(session, extension_id=EXT_B, key="k3")
+    results = ExtensionStorageDAO.list_global(EXT_A)
+    keys = {r.key for r in results}
+    assert keys == {"k1", "k2"}
+
+
+def test_list_global_filtered_by_category(session: Session) -> None:
+    _make_entry(session, key="k1", category="notebook")
+    _make_entry(session, key="k2", category="chart")
+    results = ExtensionStorageDAO.list_global(EXT_A, category="notebook")
+    assert all(r.category == "notebook" for r in results)
+    assert {r.key for r in results} == {"k1"}
+
+
+# ── User scope ────────────────────────────────────────────────────────────────
+
+
+def test_set_and_get_user(session: Session) -> None:
+    ExtensionStorageDAO.set(EXT_A, "prefs", b"data", user_fk=USER_1)
+    val = ExtensionStorageDAO.get_value(EXT_A, "prefs", user_fk=USER_1)
+    assert val == b"data"
+
+
+def test_user_scope_isolated_between_users(session: Session) -> None:
+    ExtensionStorageDAO.set(EXT_A, "prefs", b"u1", user_fk=USER_1)
+    ExtensionStorageDAO.set(EXT_A, "prefs", b"u2", user_fk=USER_2)
+    assert ExtensionStorageDAO.get_value(EXT_A, "prefs", user_fk=USER_1) == b"u1"
+    assert ExtensionStorageDAO.get_value(EXT_A, "prefs", user_fk=USER_2) == b"u2"
+
+
+def test_user_upsert(session: Session) -> None:
+    ExtensionStorageDAO.set(EXT_A, "prefs", b"old", user_fk=USER_1)
+    ExtensionStorageDAO.set(EXT_A, "prefs", b"new", user_fk=USER_1)
+    entries = ExtensionStorageDAO.list_user(EXT_A, USER_1)
+    assert len(entries) == 1
+    assert entries[0].value == b"new"
+
+
+def test_list_user_does_not_include_other_users(session: Session) -> None:
+    _make_entry(session, key="u1k", user_fk=USER_1)
+    _make_entry(session, key="u2k", user_fk=USER_2)
+    results = ExtensionStorageDAO.list_user(EXT_A, USER_1)
+    assert all(r.user_fk == USER_1 for r in results)
+
+
+# ── Resource scope ────────────────────────────────────────────────────────────
+
+
+def test_set_and_get_resource(session: Session) -> None:
+    ExtensionStorageDAO.set(
+        EXT_A,
+        "state",
+        b"res-val",
+        resource_type=RES_TYPE,
+        resource_uuid=RES_UUID,
+    )
+    val = ExtensionStorageDAO.get_value(
+        EXT_A, "state", resource_type=RES_TYPE, resource_uuid=RES_UUID
+    )
+    assert val == b"res-val"
+
+
+def test_resource_scope_isolated_by_uuid(session: Session) -> None:
+    uuid2 = "11111111-2222-3333-4444-555555555555"
+    ExtensionStorageDAO.set(
+        EXT_A, "k", b"r1", resource_type=RES_TYPE, resource_uuid=RES_UUID
+    )
+    ExtensionStorageDAO.set(
+        EXT_A, "k", b"r2", resource_type=RES_TYPE, resource_uuid=uuid2
+    )
+    assert (
+        ExtensionStorageDAO.get_value(
+            EXT_A, "k", resource_type=RES_TYPE, resource_uuid=RES_UUID
+        )
+        == b"r1"
+    )
+    assert (
+        ExtensionStorageDAO.get_value(
+            EXT_A, "k", resource_type=RES_TYPE, resource_uuid=uuid2
+        )
+        == b"r2"
+    )
+
+
+def test_list_resource(session: Session) -> None:
+    _make_entry(session, key="ra", resource_type=RES_TYPE, resource_uuid=RES_UUID)
+    _make_entry(session, key="rb", resource_type=RES_TYPE, resource_uuid=RES_UUID)
+    _make_entry(session, key="rc", resource_type="chart", resource_uuid=RES_UUID)
+    results = ExtensionStorageDAO.list_resource(EXT_A, RES_TYPE, RES_UUID)
+    keys = {r.key for r in results}
+    assert keys == {"ra", "rb"}
+
+
+def test_list_resource_filtered_by_category(session: Session) -> None:
+    _make_entry(
+        session,
+        key="r1",
+        resource_type=RES_TYPE,
+        resource_uuid=RES_UUID,
+        category="state",
+    )
+    _make_entry(
+        session,
+        key="r2",
+        resource_type=RES_TYPE,
+        resource_uuid=RES_UUID,
+        category="config",
+    )
+    results = ExtensionStorageDAO.list_resource(
+        EXT_A, RES_TYPE, RES_UUID, category="state"
+    )
+    assert len(results) == 1
+    assert results[0].key == "r1"
+
+
+# ── Pagination ────────────────────────────────────────────────────────────────
+
+
+def test_list_user_pagination(session: Session) -> None:
+    """list_user returns all rows; callers slice for pagination."""
+    for i in range(10):
+        _make_entry(session, key=f"page-{i}", user_fk=USER_1)
+    all_entries = ExtensionStorageDAO.list_user(EXT_A, USER_1)
+    assert len(all_entries) == 10
+    # Simulate first page (page_size=3)
+    page0 = all_entries[0:3]
+    page1 = all_entries[3:6]
+    assert len(page0) == 3
+    assert len(page1) == 3
+    # No overlap between pages
+    assert {e.key for e in page0}.isdisjoint({e.key for e in page1})
+
+
+def test_list_global_pagination(session: Session) -> None:
+    for i in range(5):
+        _make_entry(session, key=f"g-{i}")
+    all_entries = ExtensionStorageDAO.list_global(EXT_A)
+    assert len(all_entries) == 5
+    page = all_entries[0:2]
+    assert len(page) == 2
+
+
+def test_list_resource_pagination(session: Session) -> None:
+    for i in range(4):
+        _make_entry(
+            session, key=f"r-{i}", resource_type=RES_TYPE, resource_uuid=RES_UUID
+        )
+    all_entries = ExtensionStorageDAO.list_resource(EXT_A, RES_TYPE, RES_UUID)
+    assert len(all_entries) == 4
+
+
+# ── Encryption ────────────────────────────────────────────────────────────────
+
+
+def test_encrypted_roundtrip(session: Session) -> None:
+    """Values stored with is_encrypted=True survive a set → get_value roundtrip."""
+    secret = b"supersecret"
+    ExtensionStorageDAO.set(EXT_A, "secret_key", secret, is_encrypted=True)
+    entry = ExtensionStorageDAO.get(EXT_A, "secret_key")
+    assert entry is not None
+    assert entry.is_encrypted is True
+    # Raw stored bytes must differ from the plaintext
+    assert entry.value != secret
+    # get_value must decrypt transparently
+    assert ExtensionStorageDAO.get_value(EXT_A, "secret_key") == secret
+
+
+def test_encrypted_upsert_preserves_encryption(session: Session) -> None:
+    """Upserting an encrypted value keeps is_encrypted and re-encrypts the new value."""
+    ExtensionStorageDAO.set(EXT_A, "enc", b"v1", is_encrypted=True)
+    ExtensionStorageDAO.set(EXT_A, "enc", b"v2", is_encrypted=True)
+    assert ExtensionStorageDAO.get_value(EXT_A, "enc") == b"v2"
+
+
+def test_unencrypted_and_encrypted_keys_coexist(session: Session) -> None:
+    """Encrypted and plaintext entries in the same scope do not interfere."""
+    ExtensionStorageDAO.set(EXT_A, "plain", b"hello")
+    ExtensionStorageDAO.set(EXT_A, "secret", b"world", is_encrypted=True)
+    assert ExtensionStorageDAO.get_value(EXT_A, "plain") == b"hello"
+    assert ExtensionStorageDAO.get_value(EXT_A, "secret") == b"world"
+    plain_entry = ExtensionStorageDAO.get(EXT_A, "plain")
+    assert plain_entry is not None
+    assert plain_entry.is_encrypted is False
+    assert plain_entry.value == b"hello"
+
+
+def test_encrypted_user_scoped(session: Session) -> None:
+    """Encryption works for user-scoped entries."""
+    payload = b'{"token":"abc123"}'
+    ExtensionStorageDAO.set(EXT_A, "auth", payload, user_fk=USER_1, is_encrypted=True)
+    assert ExtensionStorageDAO.get_value(EXT_A, "auth", user_fk=USER_1) == payload
+    raw = ExtensionStorageDAO.get(EXT_A, "auth", user_fk=USER_1)
+    assert raw is not None
+    assert raw.value != payload  # stored ciphertext differs from plaintext
+
+
+def test_encrypted_value_is_fernet_token(session: Session) -> None:
+    """The stored bytes are a valid Fernet token (starts with 'gAAA' in base64)."""
+    import base64
+
+    ExtensionStorageDAO.set(EXT_A, "fernet_check", b"data", is_encrypted=True)
+    entry = ExtensionStorageDAO.get(EXT_A, "fernet_check")
+    assert entry is not None
+    # Fernet tokens are URL-safe base64 and always start with version byte 0x80
+    decoded = base64.urlsafe_b64decode(entry.value + b"==")
+    assert decoded[0] == 0x80, "stored value should be a Fernet token"
+
+
+# ── Key rotation ──────────────────────────────────────────────────────────────
+
+
+def test_key_rotation_roundtrip(session: Session, app: object) -> None:
+    """MultiFernet.rotate() re-encrypts so only the new key is needed afterwards."""
+    import base64
+    import hashlib
+
+    from cryptography.fernet import Fernet, MultiFernet
+    from flask import current_app
+
+    # Store a value encrypted with the current key (derived from SECRET_KEY).
+    payload = b"sensitive-data"
+    ExtensionStorageDAO.set(EXT_A, "rot_key", payload, is_encrypted=True)
+
+    entry = ExtensionStorageDAO.get(EXT_A, "rot_key")
+    assert entry is not None
+    old_ciphertext = entry.value
+
+    # Simulate adding a NEW key at the front of EXTENSION_STORAGE_ENCRYPTION_KEYS.
+    new_raw = b"brand-new-secret-key-32-bytes!!!"
+    new_fernet = Fernet(base64.urlsafe_b64encode(hashlib.sha256(new_raw).digest()))
+    old_raw = current_app.config["SECRET_KEY"]
+    if isinstance(old_raw, str):
+        old_raw = old_raw.encode()
+    old_fernet = Fernet(base64.urlsafe_b64encode(hashlib.sha256(old_raw).digest()))
+
+    multi = MultiFernet([new_fernet, old_fernet])
+
+    # Rotate: re-encrypt the old ciphertext with the new key.
+    entry.value = multi.rotate(old_ciphertext)
+    session.flush()
+
+    # Now the rotated ciphertext must differ from the old one…
+    assert entry.value != old_ciphertext
+    # …but decrypting with only the NEW key must return the original plaintext.
+    assert new_fernet.decrypt(entry.value) == payload
+
+
+def test_multi_fernet_decrypts_old_key(session: Session, app: object) -> None:
+    """MultiFernet tries all keys: a token encrypted with key-A is still readable
+    when key-B is current (before rotation completes)."""
+    import base64
+    import hashlib
+
+    from cryptography.fernet import Fernet, MultiFernet
+    from flask import current_app
+
+    payload = b"still-readable"
+    ExtensionStorageDAO.set(EXT_A, "multi_key", payload, is_encrypted=True)
+
+    entry = ExtensionStorageDAO.get(EXT_A, "multi_key")
+    assert entry is not None
+
+    # Build a MultiFernet with a new key first, then the current one.
+    new_raw = b"another-brand-new-secret-key!!!!"
+    new_fernet = Fernet(base64.urlsafe_b64encode(hashlib.sha256(new_raw).digest()))
+    old_raw = current_app.config["SECRET_KEY"]
+    if isinstance(old_raw, str):
+        old_raw = old_raw.encode()
+    old_fernet = Fernet(base64.urlsafe_b64encode(hashlib.sha256(old_raw).digest()))
+
+    multi = MultiFernet([new_fernet, old_fernet])
+
+    # Token was encrypted by old_fernet; multi should still decrypt it.
+    assert multi.decrypt(entry.value) == payload
+
+
+# ── Value blob ────────────────────────────────────────────────────────────────
+
+
+def test_binary_value_roundtrip(session: Session) -> None:
+    payload = bytes(range(256))
+    ExtensionStorageDAO.set(
+        EXT_A, "binkey", payload, value_type="application/octet-stream"
+    )
+    assert ExtensionStorageDAO.get_value(EXT_A, "binkey") == payload
+    entry = ExtensionStorageDAO.get(EXT_A, "binkey")
+    assert entry is not None
+    assert entry.value_type == "application/octet-stream"