derive keycloak oauth redirect_uri from configured base url

providers/keycloak/src/airflow/providers/keycloak/auth_manager/routes/login.py

@@ -57,11 +57,27 @@ class AuthManagerRefreshTokenExpiredException(Exception):  # type: ignore[no-red
 COOKIE_NAME_OAUTH_STATE = "_oauth_state"
 
 
+def _login_callback_url(request: Request) -> str:
+    """
+    Build the OAuth ``redirect_uri`` for the login callback.
+
+    Prefer the configured public base URL (as ``logout`` and the post-login redirect
+    already do) so the value handed to Keycloak does not depend on the request
+    ``Host``/``X-Forwarded-Host`` headers, which are attacker-controlled when the API
+    server runs with ``--proxy-headers``. Fall back to the request URL when
+    ``base_url`` is not configured.
+    """
+    base_url = conf.get("api", "base_url", fallback=None)
+    if base_url:
+        return urljoin(base_url, f"{AUTH_MANAGER_FASTAPI_APP_PREFIX}/login_callback")
+    return str(request.url_for("login_callback"))
+
+
 @login_router.get("/login")
 def login(request: Request) -> RedirectResponse:
     """Initiate the authentication."""
     client = KeycloakAuthManager.get_keycloak_client()
-    redirect_uri = request.url_for("login_callback")
+    redirect_uri = _login_callback_url(request)
     state = secrets.token_urlsafe(32)
     auth_url = client.auth_url(redirect_uri=str(redirect_uri), scope="openid", state=state)
     response = RedirectResponse(auth_url)
@@ -85,7 +101,7 @@ def login_callback(request: Request):
         return HTMLResponse("Invalid OAuth state parameter", status_code=403)
 
     client = KeycloakAuthManager.get_keycloak_client()
-    redirect_uri = request.url_for("login_callback")
+    redirect_uri = _login_callback_url(request)
 
     tokens = client.token(
         grant_type="authorization_code",

providers/keycloak/tests/unit/keycloak/auth_manager/routes/test_login.py

@@ -22,6 +22,8 @@
 
 from airflow.api_fastapi.app import AUTH_MANAGER_FASTAPI_APP_PREFIX
 
+from tests_common.test_utils.config import conf_vars
+
 
 class TestLoginRouter:
     @patch("airflow.providers.keycloak.auth_manager.routes.login.KeycloakAuthManager.get_keycloak_client")
@@ -121,6 +123,56 @@ def test_login_callback_sets_secure_cookies_behind_tls_proxy(
         assert any("_token=" in c and "Secure" in c for c in set_cookies)
         assert any("_id_token=" in c and "Secure" in c for c in set_cookies)
 
+    @conf_vars({("api", "base_url"): "https://airflow.example.com"})
+    @patch("airflow.providers.keycloak.auth_manager.routes.login.KeycloakAuthManager.get_keycloak_client")
+    def test_login_redirect_uri_uses_configured_base_url(self, mock_get_keycloak_client, client):
+        mock_keycloak_client = Mock()
+        mock_keycloak_client.auth_url.return_value = "redirect_url"
+        mock_get_keycloak_client.return_value = mock_keycloak_client
+        client.get(
+            "https://attacker.example.net" + AUTH_MANAGER_FASTAPI_APP_PREFIX + "/login",
+            follow_redirects=False,
+        )
+        redirect_uri = mock_keycloak_client.auth_url.call_args.kwargs["redirect_uri"]
+        assert (
+            redirect_uri
+            == "https://airflow.example.com" + AUTH_MANAGER_FASTAPI_APP_PREFIX + "/login_callback"
+        )
+
+    @conf_vars({("api", "base_url"): "https://airflow.example.com"})
+    @patch("airflow.providers.keycloak.auth_manager.routes.login.get_auth_manager")
+    @patch("airflow.providers.keycloak.auth_manager.routes.login.KeycloakAuthManager.get_keycloak_client")
+    def test_login_callback_redirect_uri_uses_configured_base_url(
+        self, mock_get_keycloak_client, mock_get_auth_manager, client
+    ):
+        state = "state"
+        mock_keycloak_client = Mock()
+        mock_keycloak_client.token.return_value = {
+            "access_token": "access_token",
+            "refresh_token": "refresh_token",
+            "id_token": "id_token",
+        }
+        mock_keycloak_client.userinfo.return_value = {
+            "sub": "sub",
+            "preferred_username": "preferred_username",
+        }
+        mock_get_keycloak_client.return_value = mock_keycloak_client
+        mock_auth_manager = Mock()
+        mock_get_auth_manager.return_value = mock_auth_manager
+        mock_auth_manager.generate_jwt.return_value = "token"
+        client.get(
+            "https://attacker.example.net"
+            + AUTH_MANAGER_FASTAPI_APP_PREFIX
+            + f"/login_callback?code=code&state={state}",
+            follow_redirects=False,
+            cookies={"_oauth_state": state},
+        )
+        redirect_uri = mock_keycloak_client.token.call_args.kwargs["redirect_uri"]
+        assert (
+            redirect_uri
+            == "https://airflow.example.com" + AUTH_MANAGER_FASTAPI_APP_PREFIX + "/login_callback"
+        )
+
     def test_login_callback_without_code(self, client):
         response = client.get(AUTH_MANAGER_FASTAPI_APP_PREFIX + "/login_callback")
         assert response.status_code == 400