Redirect to index when user does not have permission to access a page (…

…apache#36623)

airflow/www/auth.py

@@ -22,7 +22,7 @@
 from functools import wraps
 from typing import TYPE_CHECKING, Callable, Sequence, TypeVar, cast
 
-from flask import flash, redirect, render_template, request
+from flask import flash, redirect, render_template, request, url_for
 from flask_appbuilder._compat import as_unicode
 from flask_appbuilder.const import (
     FLAMSG_ERR_SEC_ACCESS_DENIED,
@@ -176,10 +176,12 @@ def _has_access(*, is_authorized: bool, func: Callable, args, kwargs):
             ),
             403,
         )
+    elif not get_auth_manager().is_logged_in():
+        return redirect(get_auth_manager().get_url_login(next=request.url))
     else:
         access_denied = get_access_denied_message()
         flash(access_denied, "danger")
-    return redirect(get_auth_manager().get_url_login(next=request.url))
+    return redirect(url_for("Airflow.index"))
 
 
 def has_access_configuration(method: ResourceMethod) -> Callable[[T], T]:

tests/www/test_auth.py

@@ -222,7 +222,23 @@ def test_has_access_dag_entities_when_unauthorized(self, mock_get_auth_manager,
             result = auth.has_access_dag_entities("GET", dag_access_entity)(self.method_test)(None, items)
 
         mock_call.assert_not_called()
-        assert result.status_code == 302
+        assert result.headers["Location"] == "/home"
+
+    @pytest.mark.db_test
+    @patch("airflow.www.auth.get_auth_manager")
+    def test_has_access_dag_entities_when_logged_out(self, mock_get_auth_manager, app, dag_access_entity):
+        auth_manager = Mock()
+        auth_manager.batch_is_authorized_dag.return_value = False
+        auth_manager.is_logged_in.return_value = False
+        auth_manager.get_url_login.return_value = "login_url"
+        mock_get_auth_manager.return_value = auth_manager
+        items = [Mock(dag_id="dag_1"), Mock(dag_id="dag_2")]
+
+        with app.test_request_context():
+            result = auth.has_access_dag_entities("GET", dag_access_entity)(self.method_test)(None, items)
+
+        mock_call.assert_not_called()
+        assert result.headers["Location"] == "login_url"
 
 
 @pytest.mark.db_test