feat: fix OAuth redirect_uri hint for proxied/iframe deployments

mcpgateway/main.py

@@ -70,6 +70,8 @@
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
 
 # First-Party
+from mcpgateway.middleware.forwarded_host import ForwardedHostMiddleware
+
 # Import the admin routes from the new module
 from mcpgateway import __version__
 from mcpgateway import version as version_module
@@ -3215,6 +3217,18 @@ async def _call_streamable_http(self, scope, receive, send):
 # This ensures all /admin/* routes (except login/logout) require admin status
 app.add_middleware(AdminAuthMiddleware)
 
+# Rewrite Host header from X-Forwarded-Host when behind a reverse proxy.
+# Uvicorn's ProxyHeadersMiddleware handles X-Forwarded-Proto and X-Forwarded-For
+# but not X-Forwarded-Host (upstream issue encode/uvicorn#965).
+# This ensures request.base_url reflects the proxy's public host, fixing the
+# OAuth redirect_uri hint and other URL construction throughout the admin UI.
+# Only registered when proxy headers are trusted (same condition as below).
+#
+# Registered BEFORE ProxyHeadersMiddleware so that it is inner (executes after
+# ProxyHeadersMiddleware in the ASGI call chain) and can rely on the scheme
+# already being corrected when deriving the default port for scope["server"].
+app.add_middleware(ForwardedHostMiddleware)
+
 # Trust all proxies (or lock down with a list of host patterns)
 app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*")
 

mcpgateway/middleware/forwarded_host.py

@@ -0,0 +1,108 @@
+# -*- coding: utf-8 -*-
+"""Location: ./mcpgateway/middleware/forwarded_host.py
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: ContextForge Contributors
+
+Forwarded Host Middleware.
+
+Rewrites the ASGI ``host`` header and ``scope["server"]`` tuple from the
+``X-Forwarded-Host`` header set by a reverse proxy.
+
+Uvicorn's ``ProxyHeadersMiddleware`` handles ``X-Forwarded-Proto`` (scheme)
+and ``X-Forwarded-For`` (client IP) but does **not** handle
+``X-Forwarded-Host`` (upstream issue encode/uvicorn#965, open PR #2811).
+
+This middleware fills that gap so that ``request.base_url`` (used in admin UI
+hints, OAuth redirect_uri display, well-known URLs, etc.) reflects the
+proxy's public host rather than the gateway's internal address.
+
+Starlette builds ``request.base_url`` from the ``host`` header, not from
+``scope["server"]``.  The host header rewrite is therefore the critical
+change; ``scope["server"]`` is updated as well for other ASGI consumers.
+
+Register this middleware **before** ``ProxyHeadersMiddleware`` in the
+``add_middleware`` stack (which means it is inner / executes **after**
+``ProxyHeadersMiddleware`` in the ASGI call chain, ensuring the scheme is
+already corrected when we derive the default port for ``scope["server"]``).
+
+Trust decisions (which upstream IPs may set forwarded headers) are the
+responsibility of the caller — this middleware always acts when
+``X-Forwarded-Host`` is present.  The gateway should only register it
+when proxy headers are trusted (the same condition under which
+``ProxyHeadersMiddleware`` is registered).
+
+When Uvicorn merges upstream support, this middleware can be removed.
+"""
+
+# Future
+from __future__ import annotations
+
+# Standard
+import logging
+from typing import Any, Awaitable, Callable, MutableMapping
+
+logger = logging.getLogger(__name__)
+
+
+class ForwardedHostMiddleware:
+    """Rewrite the ASGI ``host`` header from ``X-Forwarded-Host``.
+
+    Mirrors the approach in Uvicorn PR #2811:
+    * Parses host and optional port from the header value.
+    * Updates ``scope["server"]`` with ``(host, port)``.
+    * Replaces the ``host`` entry in ``scope["headers"]`` so that
+      Starlette's ``request.base_url`` returns the proxy origin.
+
+    Proxies typically send just the hostname for standard ports
+    (``X-Forwarded-Host: example.com``) and include the port only for
+    non-standard ones (``X-Forwarded-Host: example.com:8443``).  When
+    no port is present, ``scope["server"]`` is filled with the standard
+    default for the scheme (80 for http/ws, 443 for https/wss).
+    """
+
+    def __init__(self, app: Callable[..., Awaitable[None]]) -> None:
+        """Initialise middleware with the inner ASGI app."""
+        self.app = app
+
+    async def __call__(
+        self,
+        scope: MutableMapping[str, Any],
+        receive: Callable[[], Awaitable[MutableMapping[str, Any]]],
+        send: Callable[[MutableMapping[str, Any]], Awaitable[None]],
+    ) -> None:
+        """Rewrite host header from X-Forwarded-Host if present."""
+        if scope["type"] in ("http", "websocket"):
+            headers = dict(scope["headers"])  # type: ignore[arg-type]
+
+            if b"x-forwarded-host" in headers:
+                raw = headers[b"x-forwarded-host"].decode("latin1")
+                # Take only the first value if comma-separated (leftmost =
+                # client-facing hop).
+                x_forwarded_host = raw.split(",")[0].strip()
+
+                if x_forwarded_host:
+                    # Default port for scope["server"] when the header omits one.
+                    default_port = 443 if scope.get("scheme") in ("https", "wss") else 80
+
+                    # Parse host and optional port.
+                    # IPv6 addresses are bracketed, e.g. [::1]:8080.  A trailing
+                    # "]" means IPv6 *without* a port suffix.
+                    if ":" in x_forwarded_host and not x_forwarded_host.endswith("]"):
+                        host_part, port_str = x_forwarded_host.rsplit(":", 1)
+                        try:
+                            port = int(port_str)
+                        except ValueError:
+                            port = default_port
+                    else:
+                        host_part = x_forwarded_host
+                        port = default_port
+
+                    scope["server"] = (host_part, port)
+
+                    # Replace the ``host`` header so Starlette sees the proxy host.
+                    new_headers: list[tuple[bytes, bytes]] = [(name, value) for name, value in scope["headers"] if name != b"host"]  # type: ignore[union-attr]
+                    new_headers.append((b"host", x_forwarded_host.encode("latin1")))
+                    scope["headers"] = new_headers  # type: ignore[typeddict-item]
+
+        return await self.app(scope, receive, send)

tests/playwright/test_forwarded_host_redirect_hint.py

@@ -0,0 +1,125 @@
+# -*- coding: utf-8 -*-
+"""Location: ./tests/playwright/test_forwarded_host_redirect_hint.py
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: ContextForge Contributors
+
+Playwright tests verifying that the OAuth redirect_uri hint in the admin UI
+reflects the ``X-Forwarded-Host`` header when the gateway is behind a proxy.
+
+Covers issue #4354: the admin UI's "Use: ..." hint for the OAuth redirect_uri
+should show the proxy's public host, not the gateway's internal address.
+"""
+
+# Standard
+import re
+
+# Third-Party
+from playwright.sync_api import expect, Page
+from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
+import pytest
+
+# Local
+from .conftest import _ensure_admin_logged_in
+
+_FORWARDED_HOST = "frontend-proxy.example.com"
+_FORWARDED_PROTO = "https"
+
+# CSS selector for the OAuth redirect_uri hint <code> element in the
+# add-gateway form.  The hint lives inside #oauth-auth-code-fields-gw.
+_HINT_CODE_SELECTOR_GW = "#oauth-auth-code-fields-gw p.text-blue-600 code.bg-blue-100"
+
+
+def _wait_for_admin_content(page: Page) -> None:
+    """Wait for admin app to settle after navigation."""
+    try:
+        page.wait_for_load_state("domcontentloaded", timeout=10000)
+    except PlaywrightTimeoutError:
+        pass
+    page.wait_for_timeout(1500)
+
+
+@pytest.mark.ui
+class TestForwardedHostRedirectHint:
+    """Verify the OAuth redirect_uri hint reflects X-Forwarded-Host.
+
+    Uses ``page.route()`` to intercept requests to the admin panel and
+    re-fetch them with ``X-Forwarded-Host`` / ``X-Forwarded-Proto`` headers
+    injected, simulating a reverse-proxy deployment.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _inject_forwarded_headers(self, page: Page, base_url: str):
+        """Intercept admin requests and inject X-Forwarded-Host/Proto headers."""
+        page.set_default_timeout(10000)
+
+        def _handle_route(route):
+            # Skip SSE event-stream requests -- they are long-lived and will
+            # cause TargetClosedError on teardown.
+            if "/events" in route.request.url:
+                route.continue_()
+                return
+            response = route.fetch(
+                headers={
+                    **route.request.headers,
+                    "x-forwarded-host": _FORWARDED_HOST,
+                    "x-forwarded-proto": _FORWARDED_PROTO,
+                },
+            )
+            route.fulfill(response=response)
+
+        pattern = re.compile(re.escape(base_url.rstrip("/")) + r"/.*")
+        page.route(pattern, _handle_route)
+        yield
+        page.unroute_all(behavior="ignoreErrors")
+
+    def test_gateway_add_form_redirect_hint_uses_forwarded_host(self, page: Page, base_url: str):
+        """The redirect_uri hint in the Add Gateway form shows the proxy host."""
+        _ensure_admin_logged_in(page, base_url)
+
+        page.goto(f"{base_url.rstrip('/')}/admin/#gateways")
+        _wait_for_admin_content(page)
+
+        # Reveal OAuth fields: select auth type "oauth".
+        auth_type_select = page.locator("#auth-type-gw")
+        auth_type_select.select_option("oauth")
+        page.wait_for_selector("#auth-oauth-fields-gw", state="visible", timeout=5000)
+
+        # authorization_code is the default grant type, so
+        # #oauth-auth-code-fields-gw should already be visible.
+        page.wait_for_selector("#oauth-auth-code-fields-gw", state="visible", timeout=5000)
+
+        # Assert the hint <code> element contains the forwarded host.
+        hint_code = page.locator(_HINT_CODE_SELECTOR_GW)
+        expect(hint_code).to_be_visible()
+
+        hint_text = hint_code.text_content()
+        assert hint_text is not None, "Hint <code> element has no text content"
+
+        expected_prefix = f"{_FORWARDED_PROTO}://{_FORWARDED_HOST}"
+        assert hint_text.startswith(expected_prefix), f"Expected redirect_uri hint to start with '{expected_prefix}', but got: '{hint_text}'"
+        assert hint_text.endswith("oauth/callback"), f"Expected redirect_uri hint to end with 'oauth/callback', but got: '{hint_text}'"
+
+    def test_gateway_add_form_redirect_hint_without_forwarded_host(self, page: Page, base_url: str):
+        """Without X-Forwarded-Host, the hint shows the direct server address."""
+        # Remove the autouse routes so no forwarded headers are injected.
+        page.unroute_all(behavior="ignoreErrors")
+
+        _ensure_admin_logged_in(page, base_url)
+        page.goto(f"{base_url.rstrip('/')}/admin/#gateways")
+        _wait_for_admin_content(page)
+
+        auth_type_select = page.locator("#auth-type-gw")
+        auth_type_select.select_option("oauth")
+        page.wait_for_selector("#auth-oauth-fields-gw", state="visible", timeout=5000)
+        page.wait_for_selector("#oauth-auth-code-fields-gw", state="visible", timeout=5000)
+
+        hint_code = page.locator(_HINT_CODE_SELECTOR_GW)
+        expect(hint_code).to_be_visible()
+
+        hint_text = hint_code.text_content()
+        assert hint_text is not None, "Hint <code> element has no text content"
+
+        # Without forwarded headers, should NOT show the proxy host.
+        assert _FORWARDED_HOST not in hint_text, f"Hint should not contain forwarded host '{_FORWARDED_HOST}' without forwarding headers, but got: '{hint_text}'"
+        assert hint_text.endswith("oauth/callback"), f"Expected redirect_uri hint to end with 'oauth/callback', but got: '{hint_text}'"