fix(lang): patch FAB's LocaleView to redirect to previous page

superset/initialization/__init__.py

@@ -24,9 +24,10 @@
 
 import wtforms_json
 from deprecation import deprecated
-from flask import Flask, redirect
+from flask import abort, Flask, redirect, request, session
 from flask_appbuilder import expose, IndexView
-from flask_babel import gettext as __
+from flask_appbuilder.api import safe
+from flask_babel import gettext as __, refresh
 from flask_compress import Compress
 from flask_session import Session
 from werkzeug.middleware.proxy_fix import ProxyFix
@@ -60,6 +61,7 @@
 from superset.utils.core import is_test, pessimistic_connection_handling
 from superset.utils.decorators import transaction
 from superset.utils.log import DBEventLogger, get_event_logger_from_cfg_value
+from superset.utils.urls import is_safe_redirect_url
 
 if TYPE_CHECKING:
     from superset.app import SupersetApp
@@ -701,3 +703,28 @@ class SupersetIndexView(IndexView):
     @expose("/")
     def index(self) -> FlaskResponse:
         return redirect("/superset/welcome/")
+
+    @expose("/lang/<string:locale>")
+    @safe
+    def patch_flask_locale(self, locale: str) -> FlaskResponse:
+        """
+        Change user's locale and redirect back to the previous page.
+
+        Overrides FAB's babel.views.LocaleView so we can use the request
+        Referrer as the redirect target, in case our previous page was actually
+        served by the frontend (and thus not added to the session's page_history
+        stack).
+        """
+        if locale not in self.appbuilder.bm.languages:
+            abort(404, description="Locale not supported.")
+        session["locale"] = locale
+        refresh()
+        self.update_redirect()
+
+        redirect_to = request.headers.get("Referer")
+        if redirect_to and is_safe_redirect_url(
+            source_url=request.host_url,
+            target_url=redirect_to,
+        ):
+            return redirect(redirect_to)
+        return redirect(self.get_redirect())

superset/utils/urls.py

@@ -62,3 +62,20 @@ def is_secure_url(url: str) -> bool:
     """
     parsed_url = urlparse(url)
     return parsed_url.scheme == "https"
+
+
+def is_safe_redirect_url(source_url: str, target_url: str) -> bool:
+    """
+    Validates whether it's safe to redirect from source URL to the target URL.
+
+    Checks that the URL scheme and netloc match.
+
+    :param source_url: the current request.host_url.
+    :param target_url: the URL we plan to redirect to.
+    """
+    source_parsed = urlparse(source_url)
+    target_parsed = urlparse(target_url)
+    return (
+        source_parsed.scheme == target_parsed.scheme
+        and source_parsed.netloc == target_parsed.netloc
+    )

tests/integration_tests/core_tests.py

@@ -1223,5 +1223,56 @@ def test_dashboard_permalink(self, get_dashboard_permalink_mock, request_mock):
         assert resp.status_code == 302
 
 
+class TestLocalePatch(SupersetTestCase):
+    MOCK_LANGUAGES = (
+        "superset.views.filters.current_app.config",
+        {
+            "LANGUAGES": {
+                "es": {"flag": "es", "name": "Español"},
+            },
+        },
+    )
+
+    @mock.patch.dict(*MOCK_LANGUAGES)
+    def test_lang_redirect(self):
+        self.login(GAMMA_USERNAME)
+        referer_url = "http://localhost/explore/"
+        resp = self.client.get("/lang/es", headers={"Referer": referer_url})
+
+        assert resp.status_code == 302
+        assert resp.headers["Location"] == referer_url
+        with self.client.session_transaction() as session:
+            assert session["locale"] == "es"
+
+    @mock.patch.dict(*MOCK_LANGUAGES)
+    def test_lang_invalid_referer(self):
+        self.login(GAMMA_USERNAME)
+        referer_url = "http://someotherserver/explore/"
+        resp = self.client.get("/lang/es", headers={"Referer": referer_url})
+
+        assert resp.status_code == 302
+        assert resp.headers["Location"] == "/"
+        with self.client.session_transaction() as session:
+            assert session["locale"] == "es"
+
+    @mock.patch.dict(*MOCK_LANGUAGES)
+    def test_lang_no_referer(self):
+        self.login(GAMMA_USERNAME)
+        resp = self.client.get("/lang/es")
+
+        assert resp.status_code == 302
+        assert resp.headers["Location"] == "/"
+        with self.client.session_transaction() as session:
+            assert session["locale"] == "es"
+
+    def test_lang_invalid_locale(self):
+        self.login(GAMMA_USERNAME)
+        resp = self.client.get("/lang/es")
+
+        assert resp.status_code == 500
+        with self.client.session_transaction() as session:
+            assert session["locale"] == "en"
+
+
 if __name__ == "__main__":
     unittest.main()

tests/integration_tests/security_tests.py

@@ -1543,6 +1543,7 @@ def test_views_are_secured(self):
             ["SecurityApi", "login"],
             ["SecurityApi", "refresh"],
             ["SupersetIndexView", "index"],
+            ["SupersetIndexView", "patch_flask_locale"],
             ["DatabaseRestApi", "oauth2"],
         ]
         unsecured_views = []