AIP-84 Add safe url helper method (#47577)

* AIP-84 Add safe url helper method

* Following code review

airflow/api_fastapi/core_api/routes/public/login.py

@@ -16,11 +16,12 @@
 # under the License.
 from __future__ import annotations
 
-from fastapi import Request, status
+from fastapi import HTTPException, Request, status
 from fastapi.responses import RedirectResponse
 
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.security import is_safe_url
 
 login_router = AirflowRouter(tags=["Login"], prefix="/login")
 
@@ -33,6 +34,9 @@ def login(request: Request, next: None | str = None) -> RedirectResponse:
     """Redirect to the login URL depending on the AuthManager configured."""
     login_url = request.app.state.auth_manager.get_url_login()
 
+    if next and not is_safe_url(next):
+        raise HTTPException(status_code=400, detail="Invalid or unsafe next URL")
+
     if next:
         login_url += f"?next={next}"
     return RedirectResponse(login_url)

airflow/api_fastapi/core_api/security.py

@@ -17,7 +17,9 @@
 from __future__ import annotations
 
 from functools import cache
+from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, Callable
+from urllib.parse import urljoin, urlparse
 
 from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import OAuth2PasswordBearer
@@ -253,3 +255,23 @@ def _requires_access(
 ) -> None:
     if not is_authorized_callback():
         raise HTTPException(status.HTTP_403_FORBIDDEN, "Forbidden")
+
+
+def is_safe_url(target_url: str) -> bool:
+    """
+    Check that the URL is safe.
+
+    Needs to belong to the same domain as base_url, use HTTP or HTTPS (no JavaScript/data schemes),
+    is a valid normalized path.
+    """
+    base_url = conf.get("api", "base_url")
+
+    parsed_base = urlparse(base_url)
+    parsed_target = urlparse(urljoin(base_url, target_url))  # Resolves relative URLs
+
+    target_path = Path(parsed_target.path).resolve()
+
+    if target_path and parsed_base.path and not target_path.is_relative_to(parsed_base.path):
+        return False
+
+    return parsed_target.scheme in {"http", "https"} and parsed_target.netloc == parsed_base.netloc

tests/api_fastapi/core_api/routes/public/test_login.py

@@ -20,6 +20,8 @@
 
 import pytest
 
+from tests_common.test_utils.config import conf_vars
+
 AUTH_MANAGER_LOGIN_URL = "http://some_login_url"
 
 pytestmark = pytest.mark.db_test
@@ -39,11 +41,11 @@ class TestGetLogin(TestLoginEndpoint):
         [
             {},
             {"next": None},
-            {"next": "http://localhost:28080"},
-            {"next": "http://localhost:28080", "other_param": "something_else"},
+            {"next": "http://localhost:8080"},
+            {"next": "http://localhost:8080", "other_param": "something_else"},
         ],
     )
-    def test_should_respond_308(self, test_client, params):
+    def test_should_respond_307(self, test_client, params):
         response = test_client.get("/public/login", follow_redirects=False, params=params)
 
         assert response.status_code == 307
@@ -52,3 +54,16 @@ def test_should_respond_308(self, test_client, params):
             if params.get("next")
             else AUTH_MANAGER_LOGIN_URL
         )
+
+    @pytest.mark.parametrize(
+        "params",
+        [
+            {"next": "http://fake_domain.com:8080"},
+            {"next": "http://localhost:8080/../../up"},
+        ],
+    )
+    @conf_vars({("api", "base_url"): "http://localhost:8080/prefix"})
+    def test_should_respond_400(self, test_client, params):
+        response = test_client.get("/public/login", follow_redirects=False, params=params)
+
+        assert response.status_code == 400

tests/api_fastapi/core_api/test_security.py

@@ -25,7 +25,7 @@
 from airflow.api_fastapi.app import create_app
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
 from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser
-from airflow.api_fastapi.core_api.security import get_user, requires_access_dag
+from airflow.api_fastapi.core_api.security import get_user, is_safe_url, requires_access_dag
 
 from tests_common.test_utils.config import conf_vars
 
@@ -110,3 +110,28 @@ def test_requires_access_dag_unauthorized(self, mock_get_auth_manager):
             requires_access_dag("GET", DagAccessEntity.CODE)(fastapi_request, Mock())
 
         auth_manager.is_authorized_dag.assert_called_once()
+
+    @pytest.mark.parametrize(
+        "url, expected_is_safe",
+        [
+            ("https://server_base_url.com/prefix/some_page?with_param=3", True),
+            ("https://server_base_url.com/prefix/", True),
+            ("https://server_base_url.com/prefix", True),
+            ("/prefix/some_other", True),
+            ("prefix/some_other", True),
+            # Relative path, will go up one level escaping the prefix folder
+            ("some_other", False),
+            ("./some_other", False),
+            # wrong scheme
+            ("javascript://server_base_url.com/prefix/some_page?with_param=3", False),
+            # wrong netloc
+            ("https://some_netlock.com/prefix/some_page?with_param=3", False),
+            # Absolute path escaping the prefix folder
+            ("/some_other_page/", False),
+            # traversal, escaping the `prefix` folder
+            ("/../../../../some_page?with_param=3", False),
+        ],
+    )
+    @conf_vars({("api", "base_url"): "https://server_base_url.com/prefix"})
+    def test_is_safe_url(self, url, expected_is_safe):
+        assert is_safe_url(url) == expected_is_safe