Check for DAG ID in query param from url as well as kwargs (#32014)

Previously the dag id was only being checked in request args and form
but not kwargs, so it was possible for the id when passed as kwargs
to be None. This can allow auth for a user who does not have the
permissions to view a particular DAG.

airflow/www/auth.py

@@ -38,7 +38,8 @@ def decorated(*args, **kwargs):
             appbuilder = current_app.appbuilder
 
             dag_id = (
-                request.args.get("dag_id")
+                kwargs.get("dag_id")
+                or request.args.get("dag_id")
                 or request.form.get("dag_id")
                 or (request.is_json and request.json.get("dag_id"))
                 or None

tests/www/views/test_views_tasks.py

@@ -369,6 +369,20 @@ def test_graph_trigger_origin_graph_view(app, admin_client):
     check_content_in_response(href, resp)
 
 
+def test_graph_view_without_dag_permission(app, one_dag_perm_user_client):
+    url = "/dags/example_bash_operator/graph"
+    resp = one_dag_perm_user_client.get(url, follow_redirects=True)
+    assert resp.status_code == 200
+    assert resp.request.url == "http://localhost/dags/example_bash_operator/graph"
+    check_content_in_response("example_bash_operator", resp)
+
+    url = "/dags/example_xcom/graph"
+    resp = one_dag_perm_user_client.get(url, follow_redirects=True)
+    assert resp.status_code == 200
+    assert resp.request.url == "http://localhost/home"
+    check_content_in_response("Access is Denied", resp)
+
+
 def test_dag_details_trigger_origin_dag_details_view(app, admin_client):
     app.dag_bag.get_dag("test_graph_view").create_dagrun(
         run_type=DagRunType.SCHEDULED,
@@ -581,6 +595,39 @@ def per_dag_perm_user_client(app, new_dag_to_delete):
     delete_roles(app)
 
 
+@pytest.fixture()
+def one_dag_perm_user_client(app):
+    username = "test_user_one_dag_perm"
+    dag_id = "example_bash_operator"
+    sm = app.appbuilder.sm
+    perm = f"{permissions.RESOURCE_DAG_PREFIX}{dag_id}"
+
+    sm.create_permission(permissions.ACTION_CAN_READ, perm)
+
+    create_user(
+        app,
+        username=username,
+        role_name="User with permission to access only one dag",
+        permissions=[
+            (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_INSTANCE),
+            (permissions.ACTION_CAN_READ, permissions.RESOURCE_TASK_LOG),
+            (permissions.ACTION_CAN_READ, permissions.RESOURCE_WEBSITE),
+            (permissions.ACTION_CAN_READ, perm),
+        ],
+    )
+
+    sm.find_user(username=username)
+
+    yield client_with_login(
+        app,
+        username=username,
+        password=username,
+    )
+
+    delete_user(app, username=username)  # type: ignore
+    delete_roles(app)
+
+
 def test_delete_just_dag_per_dag_permissions(new_dag_to_delete, per_dag_perm_user_client):
     resp = per_dag_perm_user_client.post(
         f"delete?dag_id={new_dag_to_delete.dag_id}&next=/home", follow_redirects=True