Canonicalise local browser origin for Google OAuth

src/backend/server/utils_flask.py

@@ -9,6 +9,7 @@
 import threading
 import time
 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
+from urllib.parse import urlparse
 
 from flask import Flask, g, jsonify, redirect, request, url_for
 
@@ -1382,6 +1383,7 @@ def create_flask_app(
 
     _install_request_timing_middleware(app)
     _configure_flask_app_core(app, list_models_func, generate_func)
+    _install_local_browser_origin_canonicalisation(app)
     _register_default_blueprints(app)
 
     # --- Log App Version ---
@@ -2255,6 +2257,81 @@ def _log_slow_requests(response):
         return response
 
 
+_LOCAL_BROWSER_ENTRY_PATHS = frozenset(
+    {
+        "/",
+        "/von",
+        "/von/",
+        "/von/workflow-studio",
+    }
+)
+
+
+def build_browser_entry_url(bind_host: str, port: int) -> str:
+    """Build the UI URL, using the OAuth-compatible loopback hostname."""
+    browser_host = (
+        "localhost" if bind_host.strip().lower() == "127.0.0.1" else bind_host
+    )
+    return f"http://{browser_host}:{port}/von/"
+
+
+def _configured_localhost_oauth_origin() -> tuple[str, int] | None:
+    """Return an explicit localhost OAuth origin, if local canonicalisation applies."""
+    redirect_uri = os.getenv("GOOGLE_OAUTH_REDIRECT_URI", "").strip()
+    if not redirect_uri:
+        return None
+
+    try:
+        parsed = urlparse(redirect_uri)
+        if (
+            parsed.scheme.lower() != "http"
+            or (parsed.hostname or "").lower() != "localhost"
+            or parsed.username is not None
+            or parsed.password is not None
+        ):
+            return None
+        port = parsed.port or 80
+    except ValueError:
+        return None
+
+    netloc = "localhost" if port == 80 else f"localhost:{port}"
+    return f"http://{netloc}", port
+
+
+def _install_local_browser_origin_canonicalisation(app: Flask) -> None:
+    """Canonicalise only local UI entry navigation needed for Google OAuth."""
+    canonical = _configured_localhost_oauth_origin()
+    if canonical is None:
+        return
+    canonical_origin, canonical_port = canonical
+
+    @app.before_request
+    def _canonicalise_local_browser_origin():
+        if (
+            request.method not in {"GET", "HEAD"}
+            or request.path not in _LOCAL_BROWSER_ENTRY_PATHS
+            or request.scheme.lower() != "http"
+        ):
+            return None
+
+        try:
+            requested = urlparse(f"//{request.host}")
+            requested_host = (requested.hostname or "").lower()
+            requested_port = requested.port or 80
+        except ValueError:
+            return None
+
+        if requested_host != "127.0.0.1" or requested_port != canonical_port:
+            return None
+
+        target = f"{canonical_origin}{request.path}"
+        if request.query_string:
+            target = f"{target}?{request.query_string.decode('latin-1')}"
+        response = redirect(target, code=302)
+        response.headers["Cache-Control"] = "no-store"
+        return response
+
+
 def _configure_flask_app_core(
     app: Flask,
     list_models_func: Callable[[], List[str]],

src/workflows/von/main.py

@@ -19,9 +19,9 @@
 if src_root not in sys.path:
     sys.path.insert(0, src_root)
 
-create_flask_app = importlib.import_module(
-    "src.backend.server.utils_flask"
-).create_flask_app
+_utils_flask = importlib.import_module("src.backend.server.utils_flask")
+create_flask_app = _utils_flask.create_flask_app
+build_browser_entry_url = _utils_flask.build_browser_entry_url
 get_expert_tabs_enabled = importlib.import_module(
     "src.backend.services.feature_flags"
 ).get_expert_tabs_enabled
@@ -464,7 +464,7 @@ def open_browser_once_delayed():
         import requests
 
         def check_server_and_open():
-            url = f"http://{args.host}:{args.port}/von/"
+            url = build_browser_entry_url(args.host, args.port)
             for i in range(10):
                 try:
                     response = requests.get(url, timeout=1)

tests/backend/test_local_browser_origin_canonicalisation.py

@@ -0,0 +1,189 @@
+from __future__ import annotations
+
+from flask import Flask
+
+from src.backend.server import utils_flask
+
+
+def _make_app(monkeypatch, redirect_uri: str | None) -> Flask:
+    if redirect_uri is None:
+        monkeypatch.delenv("GOOGLE_OAUTH_REDIRECT_URI", raising=False)
+    else:
+        monkeypatch.setenv("GOOGLE_OAUTH_REDIRECT_URI", redirect_uri)
+
+    app = Flask(__name__)
+    app.testing = True
+    utils_flask._install_local_browser_origin_canonicalisation(app)
+
+    @app.route("/", methods=["GET", "HEAD", "POST"])
+    @app.route("/von", methods=["GET", "HEAD", "POST"])
+    @app.route("/von/", methods=["GET", "HEAD", "POST"])
+    @app.route("/von/workflow-studio", methods=["GET", "HEAD", "POST"])
+    def browser_entry():
+        return "browser-entry"
+
+    @app.route("/health")
+    def health():
+        return "healthy"
+
+    @app.route("/von/api/auth/status")
+    def auth_status():
+        return "status"
+
+    @app.route("/static/app.js")
+    def static_asset():
+        return "asset"
+
+    return app
+
+
+def test_browser_entry_url_uses_localhost_for_ipv4_loopback_bind() -> None:
+    assert (
+        utils_flask.build_browser_entry_url("127.0.0.1", 5001)
+        == "http://localhost:5001/von/"
+    )
+
+
+def test_browser_entry_url_preserves_other_bind_hosts() -> None:
+    assert (
+        utils_flask.build_browser_entry_url("localhost", 5001)
+        == "http://localhost:5001/von/"
+    )
+    assert (
+        utils_flask.build_browser_entry_url("192.0.2.10", 5001)
+        == "http://192.0.2.10:5001/von/"
+    )
+
+
+def test_loopback_browser_entry_redirects_to_configured_localhost(monkeypatch) -> None:
+    app = _make_app(
+        monkeypatch,
+        "http://localhost:5001/von/api/auth/google/callback",
+    )
+
+    response = app.test_client().get(
+        "/von/?tab=settings&return=%2Fvon%2F",
+        base_url="http://127.0.0.1:5001",
+    )
+
+    assert response.status_code == 302
+    assert (
+        response.headers["Location"]
+        == "http://localhost:5001/von/?tab=settings&return=%2Fvon%2F"
+    )
+    assert response.headers["Cache-Control"] == "no-store"
+
+
+def test_localhost_browser_entry_is_not_redirected(monkeypatch) -> None:
+    app = _make_app(
+        monkeypatch,
+        "http://localhost:5001/von/api/auth/google/callback",
+    )
+
+    response = app.test_client().get(
+        "/von/",
+        base_url="http://localhost:5001",
+    )
+
+    assert response.status_code == 200
+    assert response.get_data(as_text=True) == "browser-entry"
+
+
+def test_non_entry_loopback_requests_are_not_redirected(monkeypatch) -> None:
+    app = _make_app(
+        monkeypatch,
+        "http://localhost:5001/von/api/auth/google/callback",
+    )
+    client = app.test_client()
+
+    assert client.get("/health", base_url="http://127.0.0.1:5001").status_code == 200
+    assert (
+        client.get(
+            "/von/api/auth/status",
+            base_url="http://127.0.0.1:5001",
+        ).status_code
+        == 200
+    )
+    assert (
+        client.get(
+            "/static/app.js",
+            base_url="http://127.0.0.1:5001",
+        ).status_code
+        == 200
+    )
+
+
+def test_loopback_post_is_not_redirected(monkeypatch) -> None:
+    app = _make_app(
+        monkeypatch,
+        "http://localhost:5001/von/api/auth/google/callback",
+    )
+
+    response = app.test_client().post(
+        "/von/",
+        base_url="http://127.0.0.1:5001",
+    )
+
+    assert response.status_code == 200
+    assert response.get_data(as_text=True) == "browser-entry"
+
+
+def test_different_port_is_not_redirected(monkeypatch) -> None:
+    app = _make_app(
+        monkeypatch,
+        "http://localhost:5001/von/api/auth/google/callback",
+    )
+
+    response = app.test_client().get(
+        "/von/",
+        base_url="http://127.0.0.1:5010",
+    )
+
+    assert response.status_code == 200
+
+
+def test_hosted_or_unset_oauth_redirect_does_not_canonicalise(monkeypatch) -> None:
+    for redirect_uri in (
+        None,
+        "https://von.example.org/von/api/auth/google/callback",
+        "http://127.0.0.1:5001/von/api/auth/google/callback",
+    ):
+        app = _make_app(monkeypatch, redirect_uri)
+        response = app.test_client().get(
+            "/von/",
+            base_url="http://127.0.0.1:5001",
+        )
+        assert response.status_code == 200
+
+
+def test_von_app_factory_installs_local_browser_canonicalisation(
+    monkeypatch,
+) -> None:
+    monkeypatch.setenv(
+        "GOOGLE_OAUTH_REDIRECT_URI",
+        "http://localhost:5001/von/api/auth/google/callback",
+    )
+    monkeypatch.setenv("VON_INTERNAL_MCP_ENABLE", "0")
+    monkeypatch.setenv("VON_PREWARM_DISABLE", "1")
+    monkeypatch.setenv("VON_CONCEPT_SUMMARY_FIELD_BOOTSTRAP_ENABLE", "0")
+    monkeypatch.setenv("VON_DURABLE_WORKFLOWS_ENABLE", "0")
+    monkeypatch.setattr(utils_flask, "ensure_monitor_started", lambda: None)
+    monkeypatch.setattr(
+        utils_flask,
+        "prompt_concept_health_status",
+        lambda: {"available": True, "source_field": "stub"},
+    )
+
+    app = utils_flask.create_flask_app(
+        list_models_func=lambda: ["dummy-model"],
+        generate_func=lambda prompt, context, model: "ok",
+    )
+    app.testing = True
+
+    response = app.test_client().get(
+        "/von/?tab=settings",
+        base_url="http://127.0.0.1:5001",
+    )
+
+    assert response.status_code == 302
+    assert response.headers["Location"] == "http://localhost:5001/von/?tab=settings"