From b7b5c7e3341b2e46917cd717ced7f08cf634aa22 Mon Sep 17 00:00:00 2001
From: Niko Oliveira <onikolas@amazon.com>
Date: Mon, 19 Jun 2023 20:50:24 -0700
Subject: [PATCH] Check for DAG ID in query param from url as well as kwargs

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                 |  3 +-
 tests/www/views/test_views_tasks.py | 47 +++++++++++++++++++++++++++++
 2 files changed, 49 insertions(+), 1 deletion(-)

diff --git a/airflow/www/auth.py b/airflow/www/auth.py
index 8fdafa1f63026..0056af55fd9b3 100644
--- a/airflow/www/auth.py
+++ b/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
diff --git a/tests/www/views/test_views_tasks.py b/tests/www/views/test_views_tasks.py
index 34b58af8016bd..8ed0ba825a3e2 100644
--- a/tests/www/views/test_views_tasks.py
+++ b/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