microsoft · Chenglong-MS · May 12, 2026 · Apr 10, 2026 · Apr 10, 2026 · Apr 10, 2026
diff --git a/docs-cn/config-examples/superset/oauth_config.py b/docs-cn/config-examples/superset/oauth_config.py
@@ -15,14 +15,14 @@
 """
 
 import logging
-import json
+import os
 import secrets
 import requests
 from urllib.parse import quote, urlparse
 
 from flask_appbuilder.security.manager import AUTH_OAUTH
 from flask_appbuilder import BaseView, expose
-from flask import g, request, Response, redirect, session
+from flask import g, request, Response, redirect, render_template_string, session
 from flask_login import current_user
 from superset.security import SupersetSecurityManager
 from superset import db
@@ -170,9 +170,38 @@
 #   - /api/auth/tokens/save   (持久化到 TokenStore，供 Agent 使用)
 # =============================================================================
 
+_SSO_BRIDGE_TEMPLATE = """<!DOCTYPE html>
+<html><head><meta charset="utf-8"><title>SSO Bridge</title></head>
+<body>
+<p>正在同步登录状态...</p>
+<script nonce="{{ csp_nonce }}">
+(function() {
+    var payload = {{ payload | tojson }};
+    var targetOrigin = {{ target_origin | tojson }};
+    if (window.opener) {
+        window.opener.postMessage(payload, targetOrigin);
+        setTimeout(function() { window.close(); }, 500);
+    } else {
+        document.body.textContent = '登录成功，请关闭此窗口并返回 Data Formulator。';
+    }
+})();
+</script>
+</body></html>"""
+
+
 class SSOBridgeView(BaseView):
     route_base = "/df-sso-bridge"
 
+    # 只允许这些 DF 前端 origin 接收 Superset JWT。生产部署必须添加自己的 HTTPS origin，
+    # 也可以通过环境变量 DF_ALLOWED_ORIGINS 追加多个逗号分隔的 origin。
+    allowed_df_origins = {
+        "http://localhost:5567",
+        "http://127.0.0.1:5567",
+        "http://localhost:5173",
+        "http://127.0.0.1:5173",
+        # "https://data-formulator.example.com",
+    }
+
     @staticmethod
     def _is_real_logged_in_user():
         """区分真实登录用户和 Public 匿名用户。
@@ -194,14 +223,32 @@
         return True
 
     @staticmethod
-    def _validate_origin(raw: str) -> str:
-        """只允许 http(s) scheme，防止 javascript: 等注入。"""
-        if raw == "*":
-            return "*"
+    def _normalise_origin(raw):
+        """规范化浏览器 origin；非法或非 origin 输入返回空字符串。"""
+        raw = (raw or "").strip()
         parsed = urlparse(raw)
-        if parsed.scheme in ("http", "https") and parsed.netloc:
-            return f"{parsed.scheme}://{parsed.netloc}"
-        return "*"
+        if parsed.scheme not in ("http", "https") or not parsed.netloc:
+            return ""
+        if parsed.username or parsed.password:
+            return ""
+        if parsed.path not in ("", "/") or parsed.params or parsed.query or parsed.fragment:
+            return ""
+        return f"{parsed.scheme.lower()}://{parsed.netloc.lower()}"
+
+    @classmethod
+    def _allowed_origins(cls):
+        origins = {cls._normalise_origin(o) for o in cls.allowed_df_origins}
+        for raw in os.environ.get("DF_ALLOWED_ORIGINS", "").split(","):
+            origin = cls._normalise_origin(raw)
+            if origin:
+                origins.add(origin)
+        origins.discard("")
+        return origins
+
+    @classmethod
+    def _validate_origin(cls, raw):
+        origin = cls._normalise_origin(raw)
+        return origin if origin in cls._allowed_origins() else ""
 
     @expose("/", methods=["GET"])
     def df_sso_bridge(self):
@@ -209,8 +256,12 @@
            next_url = request.full_path.rstrip("?")
            if not next_url.startswith("/"):
                 next_url = "/"
             return redirect(f"/login/?next={quote(next_url)}")
 
+        df_origin = self._validate_origin(request.args.get("df_origin"))
+        if not df_origin:
+            return Response("Invalid df_origin", status=400, mimetype="text/plain")
+
         from flask_jwt_extended import create_access_token, create_refresh_token
 
         user_id_str = str(current_user.id)
@@ -230,32 +281,22 @@
             identity=user_id_str, additional_claims=additional_claims,
         )
 
-        df_origin = self._validate_origin(request.args.get("df_origin", "*"))
         user_data = additional_claims["user"]
         csp_nonce = getattr(g, "csp_nonce", "") or secrets.token_urlsafe(16)
 
-        html = f"""<!DOCTYPE html>
-<html><head><meta charset="utf-8"><title>SSO Bridge</title></head>
-<body>
-<p>正在同步登录状态...</p>
-<script nonce="{csp_nonce}">
-(function() {{
-    var payload = {{
-        type: 'df-sso-auth',
-        access_token: {json.dumps(access_token)},
-        refresh_token: {json.dumps(refresh_token)},
-        user: {json.dumps(user_data)}
-    }};
-    var targetOrigin = {json.dumps(df_origin)};
-    if (window.opener) {{
-        window.opener.postMessage(payload, targetOrigin);
-        setTimeout(function() {{ window.close(); }}, 500);
-    }} else {{
-        document.body.innerHTML = '<p>登录成功，请关闭此窗口并返回 Data Formulator。</p>';
-    }}
-}})();
-</script>
-</body></html>"""
+        payload = {
+            "type": "df-sso-auth",
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+            "user": user_data,
+        }
+
+        html = render_template_string(
+            _SSO_BRIDGE_TEMPLATE,
+            payload=payload,
+            target_origin=df_origin,
+            csp_nonce=csp_nonce,
+        )
         return Response(html, mimetype="text/html")
 
 

diff --git a/py-src/data_formulator/data_connector.py b/py-src/data_formulator/data_connector.py
@@ -38,6 +38,7 @@
 )
 from data_formulator.data_loader.connector_errors import classify_connector_error
 from data_formulator.datalake.parquet_utils import normalize_dtype_to_app_type
+from data_formulator.security.path_safety import ConfinedDir
 
 logger = logging.getLogger(__name__)
 
@@ -940,6 +941,11 @@ def _connectors_dir(identity: str) -> Path:
     return get_user_home(identity) / "connectors"
 
 
+def _connectors_jail(identity: str, *, mkdir: bool = True) -> ConfinedDir:
+    """Return a confined view of the per-user connectors directory."""
+    return ConfinedDir(_connectors_dir(identity), mkdir=mkdir)
+
+
 def _safe_source_filename(source_id: str) -> str:
     """Sanitise a source_id into a safe, collision-resistant filename component.
 
@@ -952,9 +958,8 @@ def _safe_source_filename(source_id: str) -> str:
 
 def _persist_user_connector(identity: str, spec: "SourceSpec") -> None:
     """Write a single connector spec to ``connectors/<source_id>.json``."""
-    cdir = _connectors_dir(identity)
-    cdir.mkdir(parents=True, exist_ok=True)
-    path = cdir / f"{_safe_source_filename(spec.source_id)}.json"
+    jail = _connectors_jail(identity)
+    path = jail.resolve(f"{_safe_source_filename(spec.source_id)}.json")
     entry = {
         "source_id": spec.source_id,
         "loader_type": spec.loader_type,
@@ -972,10 +977,12 @@ def _persist_user_connector(identity: str, spec: "SourceSpec") -> None:
 
 def _remove_user_connector(identity: str, connector_id: str) -> None:
     """Remove a connector spec from ``connectors/<source_id>.json``."""
-    path = _connectors_dir(identity) / f"{_safe_source_filename(connector_id)}.json"
     try:
+        jail = _connectors_jail(identity, mkdir=False)
+        filename = f"{_safe_source_filename(connector_id)}.json"
+        path = jail.resolve(filename)
         if path.exists():
-            path.unlink()
+            jail.unlink(filename)
             logger.info("Removed connector spec '%s'", connector_id)
     except Exception as e:
         logger.warning("Failed to remove connector spec '%s': %s", connector_id, e)

diff --git a/py-src/data_formulator/datalake/catalog_cache.py b/py-src/data_formulator/datalake/catalog_cache.py
@@ -27,6 +27,7 @@
 from typing import Any
 
 from data_formulator.datalake.naming import safe_source_id
+from data_formulator.security.path_safety import ConfinedDir
 
 logger = logging.getLogger(__name__)
 
@@ -37,8 +38,21 @@ def _cache_dir(workspace_root: Path | str) -> Path:
     return Path(workspace_root) / CATALOG_CACHE_DIR
 
 
-def _cache_file(workspace_root: Path | str, source_id: str) -> Path:
-    return _cache_dir(workspace_root) / f"{safe_source_id(source_id)}.json"
+def _cache_jail(workspace_root: Path | str, *, mkdir: bool) -> ConfinedDir:
+    return ConfinedDir(_cache_dir(workspace_root), mkdir=mkdir)
+
+
+def _cache_filename(source_id: str) -> str:
+    return f"{safe_source_id(source_id)}.json"
+
+
+def _cache_file(
+    workspace_root: Path | str,
+    source_id: str,
+    *,
+    mkdir: bool = False,
+) -> Path:
+    return _cache_jail(workspace_root, mkdir=mkdir).resolve(_cache_filename(source_id))
 
 
 def save_catalog(
@@ -48,9 +62,7 @@ def save_catalog(
 ) -> None:
     """Persist catalog data to disk. Best-effort — errors are logged, not raised."""
     try:
-        cache_dir = _cache_dir(workspace_root)
-        cache_dir.mkdir(parents=True, exist_ok=True)
-        path = _cache_file(workspace_root, source_id)
+        path = _cache_file(workspace_root, source_id, mkdir=True)
         payload = {
             "source_id": source_id,
             "synced_at": datetime.now(timezone.utc).isoformat(),
@@ -65,10 +77,11 @@ def save_catalog(
 
 def _load_catalog_raw(workspace_root: Path | str, source_id: str) -> dict[str, Any] | None:
     """Load raw catalog JSON (including original ``source_id`` key)."""
-    path = _cache_file(workspace_root, source_id)
-    if not path.exists():
-        return None
+    path: Path | None = None
     try:
+        path = _cache_file(workspace_root, source_id)
+        if not path.exists():
+            return None
         with open(path, "r", encoding="utf-8") as f:
             return json.load(f)
     except Exception:
@@ -86,13 +99,17 @@ def load_catalog(workspace_root: Path | str, source_id: str) -> list[dict[str, A
 
 def delete_catalog(workspace_root: Path | str, source_id: str) -> None:
     """Remove cached catalog file. Best-effort."""
-    path = _cache_file(workspace_root, source_id)
     try:
+        jail = _cache_jail(workspace_root, mkdir=False)
+        filename = _cache_filename(source_id)
+        path = jail.resolve(filename)
         if path.exists():
-            path.unlink()
+            jail.unlink(filename)
             logger.debug("Catalog cache deleted: %s", path)
     except Exception:
-        logger.debug("Failed to delete catalog cache %s", path, exc_info=True)
+        logger.debug(
+            "Failed to delete catalog cache for %s", source_id, exc_info=True,
+        )
 
 
 def list_cached_sources(workspace_root: Path | str) -> list[str]:

diff --git a/tests/backend/data/test_catalog_cache.py b/tests/backend/data/test_catalog_cache.py
@@ -125,6 +125,22 @@ def test_delete_removes_file(self, tmp_path: Path) -> None:
     def test_delete_nonexistent_is_silent(self, tmp_path: Path) -> None:
         delete_catalog(tmp_path, "nonexistent")
 
+    def test_delete_rejects_symlink_escape(self, tmp_path: Path) -> None:
+        cache_dir = tmp_path / "catalog_cache"
+        cache_dir.mkdir()
+        outside = tmp_path / "outside.json"
+        outside.write_text("do not delete", encoding="utf-8")
+        link = cache_dir / "pg_prod.json"
+        try:
+            link.symlink_to(outside)
+        except OSError:
+            pytest.skip("symlink creation is not available on this platform")
+
+        delete_catalog(tmp_path, "pg_prod")
+
+        assert outside.exists()
+        assert link.is_symlink()
+
 
 # ==================================================================
 # Tests: list_cached_sources

diff --git a/tests/backend/data/test_connector_directory_storage.py b/tests/backend/data/test_connector_directory_storage.py
@@ -186,6 +186,24 @@ def test_remove_nonexistent_is_silent(self, tmp_path: Path) -> None:
         with patch("data_formulator.datalake.workspace.get_user_home", return_value=user_dir):
             _remove_user_connector("alice", "nonexistent")
 
+    def test_remove_rejects_symlink_escape(self, tmp_path: Path) -> None:
+        user_dir = tmp_path / "users" / "alice"
+        connectors_dir = user_dir / "connectors"
+        connectors_dir.mkdir(parents=True)
+        outside = tmp_path / "outside.json"
+        outside.write_text("do not delete", encoding="utf-8")
+        link = connectors_dir / "mysql--prod.json"
+        try:
+            link.symlink_to(outside)
+        except OSError:
+            pytest.skip("symlink creation is not available on this platform")
+
+        with patch("data_formulator.datalake.workspace.get_user_home", return_value=user_dir):
+            _remove_user_connector("alice", "mysql:prod")
+
+        assert outside.exists()
+        assert link.is_symlink()
+
 
 # ==================================================================
 # Tests: _load_user_specs from connectors/ directory